//! Moving the working copy to another state (spec §12.3, §7.7).

use super::*;

#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct CheckoutReport {
    pub written: usize,
    pub removed: usize,
    pub conflicted: bool,
    /// Nested repositories that could not be moved, and other remarks.
    pub notes: Vec<String>,
    /// How many changes of the state left behind the new one lacks.
    pub lacks: usize,
}

/// Text files rendered at a time by a checkout: bounds its memory.
const CHECKOUT_BATCH: usize = 2048;

impl Repo {
    /// Move the working copy to the state whose heads are `heads` (spec §12.3):
    /// the working directory is rewritten to that state and a fresh
    /// working-copy change is started on it. The bookmark does not move.
    pub fn checkout(
        &self,
        heads: &[ChangeHash],
        discard: bool,
        command: &str,
    ) -> Result<CheckoutReport> {
        self.checkout_keeping(heads, discard, command, &BTreeSet::new())
    }

    /// [`Repo::checkout`], leaving the nested repositories at
    /// `keep_subrepos` as they are (their pins are carried work).
    pub(crate) fn checkout_keeping(
        &self,
        heads: &[ChangeHash],
        discard: bool,
        command: &str,
        keep_subrepos: &BTreeSet<String>,
    ) -> Result<CheckoutReport> {
        let _wc = self.lock_working_copy()?;
        let _all = crate::timing::phase("checkout");
        // A working copy clean by the manifest needs no snapshot; otherwise a
        // status snapshot, which neither reads nor renders files the
        // manifest proves unchanged, and whose base is history only
        // (oxygen#35).
        let (base, changed_paths) = if self.clean_by_manifest()? {
            let (_, wc) = self.working_copy()?;
            (self.load_state_history(&wc.deps)?, Vec::new())
        } else {
            let snap = self.snapshot_status(command)?;
            self.refuse_dirty(&snap, discard)?;
            let paths = snap.statuses.iter().map(|s| s.path.clone()).collect();
            (snap.base, paths)
        };
        let views = crate::timing::phase("checkout: manifest and base view");
        let old_manifest = self.read_manifest()?.map(|(_, _, m)| m).unwrap_or_default();
        let bview = base.tree().view();
        drop(views);
        let changed: HashSet<&str> = changed_paths.iter().map(String::as_str).collect();
        // The target's history; its file graphs are built a batch at a time
        // for the files that must be written, and dropped after.
        let mut target = self.load_state_history(heads)?;
        let heads = target.maximal(heads);
        let tv = crate::timing::phase("checkout: target view");
        let tview = target.tree().view();
        drop(tv);
        let mut report = CheckoutReport::default();
        // Paths this file system cannot hold side by side (spec §7.7).
        report.notes.extend(
            self.platform
                .problems(tview.entries.keys().map(String::as_str)),
        );
        report.conflicted = tview.is_conflicted();
        report.lacks = base.changes().filter(|(h, _)| !target.contains(h)).count();

        // Remove what the base or the working copy has and the target does
        // not. A nested repository is never deleted (spec §7.8).
        let mut dirs_to_prune: BTreeSet<PathBuf> = BTreeSet::new();
        let gone: BTreeSet<&str> = bview
            .entries
            .iter()
            .filter(|(_, f)| f.kind != NodeKind::Dir)
            .map(|(p, _)| p.as_str())
            .chain(changed.iter().copied())
            .filter(|p| !tview.entries.contains_key(*p))
            .collect();
        for path in gone {
            if bview
                .entries
                .get(path)
                .is_some_and(|f| f.kind == NodeKind::Subrepo)
            {
                report
                    .notes
                    .push(format!("{path}: nested repository left in place"));
                continue;
            }
            let full = self.safe_path(path)?;
            if fs::symlink_metadata(&full).is_err() {
                continue;
            }
            fs::remove_file(&full)?;
            report.removed += 1;
            if let Some(parent) = full.parent() {
                dirs_to_prune.insert(parent.to_path_buf());
            }
        }

        // The manifest after the checkout, built from what is known.
        let plan = crate::timing::phase("checkout: plan");
        let mut entries: BTreeMap<String, FsEntry> = BTreeMap::new();
        let hex = |b: &[u8]| blake3::hash(b).to_hex().to_string();
        // Text files to render, attribute files first (the others need them).
        let mut texts: Vec<(String, NodeId, Mode)> = Vec::new();
        for (path, e) in &tview.entries {
            if e.kind == NodeKind::Dir || !self.in_scope(path) {
                continue;
            }
            let untouched = !changed.contains(path.as_str())
                && bview
                    .entries
                    .get(path)
                    .is_some_and(|b| b.node == e.node && b.kind == e.kind && b.mode == e.mode);
            if e.kind == NodeKind::File && !e.mode.opaque {
                // The manifest proves the file on disk equal to the base's
                // rendering (its key is the base graph's, and its stat was
                // checked just now), and the graph is the same in both
                // states: nothing to read, render or write.
                let key = base.graph_key(&e.node);
                let same_graph = untouched
                    && !is_attributes(path)
                    && key.is_some()
                    && key == target.graph_key(&e.node);
                let proven = |m: &&prompt::ManifestEntry| {
                    same_graph
                        && m.graph.as_deref() == key.map(|k| ObjectRef(k).to_hex()).as_deref()
                };
                match old_manifest.entries.get(path).filter(proven) {
                    Some(m) => {
                        entries.insert(
                            path.clone(),
                            FsEntry {
                                content_hash: Some(m.hash.clone()),
                                graph_key: m.graph.clone(),
                                unchanged: true,
                                ..FsEntry::new(NodeKind::File, e.mode.exec, Vec::new())
                            },
                        );
                    }
                    None => texts.push((path.clone(), e.node, e.mode)),
                }
                continue;
            }
            // Opaque files, symlinks and subrepos: their blobs.
            let blobs = target.live_blobs(&e.node);
            if blobs.len() > 1 {
                report.conflicted = true;
            }
            let Some(blob) = blobs.first().copied() else {
                continue;
            };
            if e.kind == NodeKind::Subrepo {
                if keep_subrepos.contains(path) {
                    continue;
                }
                let pin = crate::subrepo::Pin::decode(&self.read_blob(&blob)?)?;
                let sync = self.sync_subrepo(path, &pin, false)?;
                if matches!(
                    sync,
                    SubrepoSync::Updated | SubrepoSync::KeptBookmark { .. }
                ) {
                    report.written += 1;
                }
                report.notes.extend(sync.note(path));
                continue;
            }
            let same_blob = untouched && base.live_blobs(&e.node) == blobs;
            if let Some(m) = old_manifest
                .entries
                .get(path)
                .filter(|_| same_blob && blobs.len() == 1)
            {
                entries.insert(
                    path.clone(),
                    FsEntry {
                        content_hash: Some(m.hash.clone()),
                        blob: (e.kind == NodeKind::File).then_some(blob),
                        ..FsEntry::new(e.kind, e.mode.exec, Vec::new())
                    },
                );
                continue;
            }
            let bytes = self.read_blob(&blob)?;
            let full = self.root.join(path);
            if e.kind == NodeKind::Symlink {
                let same = fs::read_link(&full)
                    .is_ok_and(|t| t.to_string_lossy().as_bytes() == bytes.as_slice());
                if !same {
                    self.write_symlink(path, &bytes)?;
                    report.written += 1;
                }
                entries.insert(
                    path.clone(),
                    FsEntry {
                        content_hash: Some(hex(&bytes)),
                        ..FsEntry::new(NodeKind::Symlink, false, Vec::new())
                    },
                );
            } else {
                if !self.same_on_disk(&full, &bytes, e.mode.exec) {
                    self.write_file(path, &bytes, e.mode.exec)?;
                    report.written += 1;
                }
                entries.insert(
                    path.clone(),
                    FsEntry {
                        content_hash: Some(hex(&bytes)),
                        blob: (blobs.len() == 1).then_some(blob),
                        ..FsEntry::new(NodeKind::File, e.mode.exec, Vec::new())
                    },
                );
            }
            for (i, b) in blobs.iter().enumerate().skip(1) {
                self.write_file(
                    &format!("{path}.oxy-conflict-{}", i + 1),
                    &self.read_blob(b)?,
                    e.mode.exec,
                )?;
            }
        }

        drop(plan);
        let _render = crate::timing::phase("checkout: render and write");
        // Render and write text files a batch at a time; attribute files
        // come first, since the others' line endings and locks follow them.
        texts.sort_by_key(|(p, _, _)| !is_attributes(p));
        let keys: HashMap<NodeId, [u8; 32]> = texts
            .iter()
            .filter_map(|(_, n, _)| Some((*n, target.graph_key(n)?)))
            .collect();
        let mut attrs = crate::attributes::Attributes::default();
        let attr_count = texts.iter().filter(|(p, _, _)| is_attributes(p)).count();
        let mut done = 0usize;
        let mut batches: Vec<&[(String, NodeId, Mode)]> = Vec::new();
        if attr_count > 0 {
            batches.push(&texts[..attr_count]);
        }
        batches.extend(texts[attr_count..].chunks(CHECKOUT_BATCH));
        for batch in batches {
            use rayon::prelude::*;
            let nodes: Vec<NodeId> = batch.iter().map(|(_, n, _)| *n).collect();
            self.load_graphs_of(&mut target, &nodes)?;
            let rendered: Vec<(Vec<u8>, bool)> = nodes
                .par_iter()
                .map(|n| {
                    let (bytes, m) = target.render_file(n);
                    (bytes, m.is_conflicted())
                })
                .collect();
            target.drop_graphs(&nodes);
            if done == 0 && attr_count > 0 {
                attrs = crate::attributes::Attributes::from_files(
                    batch
                        .iter()
                        .zip(&rendered)
                        .map(|((p, _, _), (b, _))| (p.as_str(), b.as_slice())),
                );
            }
            done += batch.len();
            for ((path, node, mode), (bytes, conflicted)) in batch.iter().zip(rendered) {
                let a = attrs.lookup(path);
                let crlf = a.eol == Some(crate::attributes::Eol::Crlf);
                let on_disk: Vec<u8> = if crlf {
                    crate::attributes::to_crlf(&bytes)
                } else {
                    bytes
                };
                if !self.same_on_disk(&self.root.join(path), &on_disk, mode.exec) {
                    self.write_file(path, &on_disk, mode.exec)?;
                    report.written += 1;
                }
                report.conflicted |= conflicted;
                entries.insert(
                    path.clone(),
                    FsEntry {
                        content_hash: Some(hex(&on_disk)),
                        graph_key: (!conflicted && a.eol.is_none())
                            .then(|| keys.get(node).map(|k| ObjectRef(*k).to_hex()))
                            .flatten(),
                        ..FsEntry::new(NodeKind::File, mode.exec, Vec::new())
                    },
                );
            }
        }
        // Lockable files are read-only until locked (spec §18), whether
        // written now or left as they were.
        for (path, e) in &entries {
            if e.kind == NodeKind::File && attrs.lookup(path).lockable && !self.is_locked(path) {
                let _ = set_readonly(&self.root.join(path), true);
            }
        }
        // Prune directories left empty, deepest first.
        for d in dirs_to_prune.iter().rev() {
            let mut d = d.clone();
            while d != self.root
                && fs::read_dir(&d)
                    .map(|mut i| i.next().is_none())
                    .unwrap_or(false)
            {
                fs::remove_dir(&d)?;
                let Some(p) = d.parent() else { break };
                d = p.to_path_buf();
            }
        }

        drop(_render);
        let _t = crate::timing::phase("checkout: record and manifest");
        let (op_ref, op) = self.head_op()?;
        let wc = self.new_working_copy(heads.clone(), "");
        let wc_hash = self.store.put_change(&wc)?;
        let mut view = op.view.clone();
        view.workspaces.insert(self.workspace, wc_hash);
        self.record_op(&[op_ref], command, view)?;
        self.write_manifest_entries(report.conflicted, false, &entries, &HashMap::new())?;
        let _ = self.run_hook(
            "post-checkout",
            &format!(
                "{{\"event\":\"post-checkout\",\"heads\":{},\"written\":{},\"removed\":{}}}",
                json_array(heads.iter().map(|h| json_str(&h.to_hex()))),
                report.written,
                report.removed
            ),
        );
        Ok(report)
    }

    /// Refuse to leave uncommitted work behind unless `discard`.
    fn refuse_dirty(&self, snap: &SnapshotResult, discard: bool) -> Result<()> {
        if discard || (snap.wc.tree_ops.is_empty() && snap.wc.edits.is_empty()) {
            return Ok(());
        }
        let list: Vec<String> = snap
            .statuses
            .iter()
            .take(8)
            .map(|s| format!("{} {}", status_word(s.status), s.path))
            .collect();
        fail(format!(
            "the working copy has changes ({}{}); commit them, take them along with --carry, or drop them with --discard",
            list.join(", "),
            if snap.statuses.len() > 8 { ", …" } else { "" }
        ))
    }

    /// Whether the file at `full` already holds `bytes` with this executable
    /// bit, so writing it again can be skipped.
    fn same_on_disk(&self, full: &Path, bytes: &[u8], exec: bool) -> bool {
        match fs::symlink_metadata(full) {
            Ok(m) if m.is_file() && m.len() == bytes.len() as u64 => {
                (!self.platform.exec_bits || is_executable(&m) == exec)
                    && fs::read(full).is_ok_and(|b| b == bytes)
            }
            _ => false,
        }
    }

    /// The absolute path for a repository path, refusing anything that would
    /// escape the workspace or write through a symlink (spec §7.7).
    fn safe_path(&self, rel: &str) -> Result<PathBuf> {
        let mut full = self.root.clone();
        let comps: Vec<&str> = rel.split('/').collect();
        for (i, comp) in comps.iter().enumerate() {
            if comp.is_empty() || *comp == "." || *comp == ".." || comp.contains('\\') {
                return fail(format!("refusing unsafe path {rel:?}"));
            }
            if i == 0 && *comp == OXY_DIR {
                return fail(format!("refusing to write inside {OXY_DIR}: {rel:?}"));
            }
            if i + 1 < comps.len() {
                full.push(comp);
                if let Ok(m) = fs::symlink_metadata(&full) {
                    if m.file_type().is_symlink() {
                        return fail(format!(
                            "refusing to write through symlink {}",
                            full.display()
                        ));
                    }
                }
            } else {
                full.push(comp);
            }
        }
        Ok(full)
    }

    pub(crate) fn write_file(&self, rel: &str, bytes: &[u8], exec: bool) -> Result<()> {
        let full = self.safe_path(rel)?;
        if let Some(parent) = full.parent() {
            fs::create_dir_all(parent)?;
        }
        if fs::symlink_metadata(&full).is_ok_and(|m| m.file_type().is_symlink() || m.is_dir()) {
            remove_any(&full)?;
        }
        if fs::metadata(&full).is_ok_and(|m| m.permissions().readonly()) {
            set_readonly(&full, false)?;
        }
        fs::write(&full, bytes)?;
        set_executable(&full, exec)?;
        Ok(())
    }

    pub(crate) fn write_symlink(&self, rel: &str, target: &[u8]) -> Result<()> {
        let full = self.safe_path(rel)?;
        if let Some(parent) = full.parent() {
            fs::create_dir_all(parent)?;
        }
        if fs::symlink_metadata(&full).is_ok() {
            remove_any(&full)?;
        }
        #[cfg(unix)]
        {
            std::os::unix::fs::symlink(String::from_utf8_lossy(target).as_ref(), &full)?;
            Ok(())
        }
        #[cfg(not(unix))]
        {
            // Spec §7.7: without symlink support, write the target as a plain file.
            fs::write(&full, target)?;
            Ok(())
        }
    }
}

/// How a refusal names a path's change: "removed" is a file of the working
/// copy's base that is missing on disk.
fn status_word(s: ChangeStatus) -> &'static str {
    match s {
        ChangeStatus::Added => "added",
        ChangeStatus::Modified => "modified",
        ChangeStatus::Removed => "removed",
        ChangeStatus::Renamed => "renamed",
        ChangeStatus::ModeChanged => "mode changed",
        ChangeStatus::ConflictEdited => "conflict edited",
        ChangeStatus::Resolved => "resolved",
    }
}