use std::collections::BTreeSet;
use std::ffi::OsStr;
use std::fs;
use std::path::{Component, Path, PathBuf};
use std::process::ExitCode;

use anyhow::{bail, Context, Result};
use serde::Serialize;

use super::link;
use crate::output::CommandReport;
use crate::paths::state::StateLayout;
use crate::profile::{self, ProfileName, DEFAULT_PROFILE};
use crate::repo::marker as repo_marker;
use crate::repo::registry::{self as repo_registry, RepoAliases, RepoRegistryEntry};

#[derive(Serialize)]
pub struct RepoStatusReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    locality_id: String,
    display_name: Option<String>,
    marker_path: String,
    registry_path: String,
    registry_present: bool,
    overlay_path: String,
    overlay_present: bool,
    current_clone_root: String,
    current_clone_registered: bool,
    detected_aliases: RepoAliases,
    stored_aliases: Option<RepoAliases>,
    alias_matches: Vec<String>,
    known_clone_roots: Vec<String>,
    warnings: Vec<String>,
}

impl CommandReport for RepoStatusReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        println!("Profile: {}", self.profile);
        println!("Project ID: {}", self.locality_id);
        if let Some(display_name) = &self.display_name {
            println!("Display name: {display_name}");
        }
        println!("Marker: {}", self.marker_path);
        println!(
            "Registry: {} ({})",
            self.registry_path,
            presence_label(self.registry_present)
        );
        println!(
            "Project overlay: {} ({})",
            self.overlay_path,
            presence_label(self.overlay_present)
        );
        println!("Current workspace: {}", self.current_clone_root);

        if self.detected_aliases.remote_urls.is_empty()
            && self.detected_aliases.root_commits.is_empty()
            && self.detected_aliases.repo_basenames.is_empty()
        {
            println!("Detected aliases: none");
        } else {
            render_aliases("Detected aliases", &self.detected_aliases);
        }

        match &self.stored_aliases {
            Some(aliases) => render_aliases("Stored alias hints", aliases),
            None => println!("Stored alias hints: none"),
        }

        if self.alias_matches.is_empty() {
            println!("Alias matches: none");
        } else {
            println!("Alias matches: {}", self.alias_matches.join(", "));
        }

        if self.known_clone_roots.is_empty() {
            println!("Known linked workspaces: none recorded");
        } else {
            println!("Known linked workspaces:");
            for clone_root in &self.known_clone_roots {
                println!("  {clone_root}");
            }
        }

        for warning in &self.warnings {
            println!("Warning: {warning}");
        }
    }
}

#[derive(Serialize)]
pub struct RepoListReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    current_locality_id: Option<String>,
    repos: Vec<RepoListEntry>,
}

impl CommandReport for RepoListReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        if self.repos.is_empty() {
            println!(
                "No project overlays registered for profile {}.",
                self.profile
            );
            return;
        }

        println!("Known project overlays for profile {}:", self.profile);
        for repo in &self.repos {
            let current = if repo.current { "*" } else { "-" };
            let display_name = repo
                .display_name
                .as_deref()
                .map(|name| format!(" ({name})"))
                .unwrap_or_default();
            println!(
                "{current} {}{} | project overlay: {} | workspaces: {}",
                repo.locality_id,
                display_name,
                presence_label(repo.overlay_present),
                repo.known_clone_roots.len()
            );
        }
    }
}

#[derive(Serialize)]
pub struct RepoListEntry {
    locality_id: String,
    display_name: Option<String>,
    registry_path: String,
    overlay_path: String,
    overlay_present: bool,
    known_clone_roots: Vec<String>,
    aliases: RepoAliases,
    current: bool,
}

#[derive(Serialize)]
pub struct RepoRelinkReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    previous_locality_id: String,
    locality_id: String,
    marker_path: String,
    registry_path: String,
    overlay_path: String,
    warnings: Vec<String>,
}

impl CommandReport for RepoRelinkReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        println!(
            "Relinked current workspace from project ID {} to {}.",
            self.previous_locality_id, self.locality_id
        );
        println!("Marker: {}", self.marker_path);
        println!("Registry: {}", self.registry_path);
        println!("Project overlay: {}", self.overlay_path);

        for warning in &self.warnings {
            println!("Warning: {warning}");
        }
    }
}

#[derive(Serialize)]
pub struct RepoMergeReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    target_locality_id: String,
    source_locality_id: String,
    marker_updated: bool,
    marker_path: String,
    registry_path: String,
    removed_registry_path: String,
    merged_overlay_paths: Vec<String>,
    warnings: Vec<String>,
}

impl CommandReport for RepoMergeReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        println!(
            "Merged project ID {} into {}.",
            self.source_locality_id, self.target_locality_id
        );
        if self.marker_updated {
            println!("Marker updated: {}", self.marker_path);
        } else {
            println!("Marker unchanged: {}", self.marker_path);
        }
        println!("Registry kept: {}", self.registry_path);
        println!("Registry removed: {}", self.removed_registry_path);

        if !self.merged_overlay_paths.is_empty() {
            println!("Merged project-overlay paths:");
            for path in &self.merged_overlay_paths {
                println!("  {path}");
            }
        }

        for warning in &self.warnings {
            println!("Warning: {warning}");
        }
    }
}

#[derive(Serialize)]
pub struct RepoSplitReport {
    command: &'static str,
    ok: bool,
    path: String,
    profile: String,
    source_locality_id: String,
    locality_id: String,
    marker_path: String,
    registry_path: String,
    overlay_path: String,
    copied_overlay_paths: Vec<String>,
    warnings: Vec<String>,
}

impl CommandReport for RepoSplitReport {
    fn exit_code(&self) -> ExitCode {
        ExitCode::SUCCESS
    }

    fn render_text(&self) {
        println!(
            "Split current workspace from project ID {} into new project ID {}.",
            self.source_locality_id, self.locality_id
        );
        println!("Marker: {}", self.marker_path);
        println!("Registry: {}", self.registry_path);
        println!("Project overlay: {}", self.overlay_path);

        if !self.copied_overlay_paths.is_empty() {
            println!("Copied project-overlay paths:");
            for path in &self.copied_overlay_paths {
                println!("  {path}");
            }
        }

        for warning in &self.warnings {
            println!("Warning: {warning}");
        }
    }
}

#[derive(Default)]
struct RepoStateChangeSummary {
    overlay_paths: Vec<String>,
}

#[derive(Clone, Copy, Eq, PartialEq)]
enum ExistingPathKind {
    Dir,
    File,
    Unsupported,
}

pub fn status(repo_root: &Path, explicit_profile: Option<&str>) -> Result<RepoStatusReport> {
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    let marker =
        repo_marker::load(repo_root)?.ok_or_else(|| linked_clone_error(repo_root, &layout))?;
    let locality_id = marker.locality_id;
    let marker_path = repo_root.join(repo_marker::MARKER_FILE);
    let registry_path = layout.repo_metadata_path(&locality_id)?;
    let overlay_path = layout.repo_overlay_root(&locality_id)?;
    let current_clone_root = repo_registry::normalize_clone_root(repo_root)?;
    let detected_aliases = link::collect_aliases(repo_root)?;
    let alias_matches = link::find_alias_match_locality_ids(&layout, &detected_aliases)?;

    let registry_entry = repo_registry::load(&registry_path)?;
    let registry_present = registry_entry.is_some();
    let overlay_present = overlay_path.is_dir();
    let display_name = registry_entry
        .as_ref()
        .and_then(|entry| entry.display_name.clone())
        .or(marker.display_name);
    let stored_aliases = registry_entry.as_ref().map(|entry| entry.aliases.clone());
    let known_clone_roots = registry_entry
        .as_ref()
        .map(|entry| entry.known_clone_roots.clone())
        .unwrap_or_default();
    let current_clone_registered = known_clone_roots
        .iter()
        .any(|root| root == &current_clone_root);

    let mut warnings = Vec::new();
    if !registry_present {
        warnings.push(format!(
            "ccd repo status found that the project registry is missing at {}; project identity is only partially recorded, so repo repair commands may miss metadata. Run `{}` to recreate it.",
            registry_path.display(),
            link_command(repo_root, &layout, Some(&locality_id))
        ));
    }
    if !overlay_present {
        warnings.push(format!(
            "ccd repo status found that the project overlay is missing at {}; project policy, memory, and runtime state for locality_id `{locality_id}` are unavailable. Run `{}` to recreate the local overlay, or `{}` to reconnect it.",
            overlay_path.display(),
            attach_command(repo_root, &layout),
            link_command(repo_root, &layout, Some(&locality_id))
        ));
    }
    if registry_present && !current_clone_registered {
        warnings.push(format!(
            "ccd repo status found that current workspace root {current_clone_root} is not recorded in the project registry; future relink, merge, or split operations may miss this clone. Run `{}` to register it again.",
            link_command(repo_root, &layout, Some(&locality_id))
        ));
    }
    if alias_matches.len() > 1 {
        warnings.push(format!(
            "multiple project IDs match the current workspace aliases: {}",
            alias_matches.join(", ")
        ));
    }

    Ok(RepoStatusReport {
        command: "repo-status",
        ok: true,
        path: repo_root.display().to_string(),
        profile: profile.to_string(),
        locality_id,
        display_name,
        marker_path: marker_path.display().to_string(),
        registry_path: registry_path.display().to_string(),
        registry_present,
        overlay_path: overlay_path.display().to_string(),
        overlay_present,
        current_clone_root,
        current_clone_registered,
        detected_aliases,
        stored_aliases,
        alias_matches,
        known_clone_roots,
        warnings,
    })
}

pub fn list(repo_root: &Path, explicit_profile: Option<&str>) -> Result<RepoListReport> {
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    let current_locality_id = repo_marker::load(repo_root)?.map(|marker| marker.locality_id);
    let repos = load_registry_entries(&layout, &current_locality_id)?;

    Ok(RepoListReport {
        command: "repo-list",
        ok: true,
        path: repo_root.display().to_string(),
        profile: profile.to_string(),
        current_locality_id,
        repos,
    })
}

pub fn relink(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    target_locality_id: &str,
) -> Result<RepoRelinkReport> {
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    let existing_marker =
        repo_marker::load(repo_root)?.ok_or_else(|| linked_clone_error(repo_root, &layout))?;
    let previous_locality_id = existing_marker.locality_id.clone();

    let registry_path = layout.repo_metadata_path(target_locality_id)?;
    if repo_registry::load(&registry_path)?.is_none() {
        bail!(
            "locality_id `{target_locality_id}` is not registered at {}; inspect known project overlays with `ccd repo list --path {}` first",
            registry_path.display(),
            repo_root.display()
        );
    }

    let result = link::link_repo(
        repo_root,
        link::LinkRequest {
            explicit_profile,
            explicit_locality_id: Some(target_locality_id),
            display_name: None,
        },
    )?;

    let mut warnings = result.warnings;
    if previous_locality_id != target_locality_id {
        warnings.push(clone_state_warning(repo_root));
    }

    Ok(RepoRelinkReport {
        command: "repo-relink",
        ok: true,
        path: repo_root.display().to_string(),
        profile: profile.to_string(),
        previous_locality_id,
        locality_id: result.locality_id,
        marker_path: result.marker_path.display().to_string(),
        registry_path: result.registry_path.display().to_string(),
        overlay_path: result.overlay_path.display().to_string(),
        warnings,
    })
}

pub fn merge(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    target_locality_id: &str,
    source_locality_id: &str,
    force: bool,
) -> Result<RepoMergeReport> {
    if !force {
        bail!(
            "repo merge is destructive; re-run with `ccd repo merge {target_locality_id} {source_locality_id} --force --path {}` to confirm",
            repo_root.display()
        );
    }
    if target_locality_id == source_locality_id {
        bail!("target_locality_id and source_locality_id must differ");
    }

    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    let target_registry_path = layout.repo_metadata_path(target_locality_id)?;
    let source_registry_path = layout.repo_metadata_path(source_locality_id)?;
    let target_entry = repo_registry::load(&target_registry_path)?.ok_or_else(|| {
        anyhow::anyhow!(
            "target locality_id `{target_locality_id}` is not registered at {}",
            target_registry_path.display()
        )
    })?;
    let source_entry = repo_registry::load(&source_registry_path)?.ok_or_else(|| {
        anyhow::anyhow!(
            "source locality_id `{source_locality_id}` is not registered at {}",
            source_registry_path.display()
        )
    })?;

    let conflicts = collect_merge_conflicts(&layout, source_locality_id, target_locality_id)
        .with_context(|| {
            format!(
                "failed to scan repo state across profiles before merging locality_id `{source_locality_id}` into `{target_locality_id}`; no changes were made"
            )
        })?;
    if !conflicts.is_empty() {
        bail!(
            "repo merge would overwrite differing state:\n- {}",
            conflicts.join("\n- ")
        );
    }

    let merged_state = merge_repo_state_across_profiles(&layout, source_locality_id, target_locality_id)
        .with_context(|| {
            format!(
                "merge failed while transferring repo state from locality_id `{source_locality_id}` into `{target_locality_id}` across profiles; registry metadata was not updated"
            )
        })?;
    let merged_entry = merge_registry_entries(target_entry, source_entry.clone()).with_context(
        || {
            format!(
                "merge transferred repo state from locality_id `{source_locality_id}` into `{target_locality_id}`, but failed to combine registry metadata"
            )
        },
    )?;
    repo_registry::write(&target_registry_path, &merged_entry).with_context(|| {
        format!(
            "merge transferred repo state from locality_id `{source_locality_id}` into `{target_locality_id}`, but failed to write the merged registry at {}",
            target_registry_path.display()
        )
    })?;

    let marker_path = repo_root.join(repo_marker::MARKER_FILE);
    let mut marker_updated = false;
    if let Some(marker) = repo_marker::load(repo_root)? {
        if marker.locality_id == source_locality_id {
            let display_name = link::choose_display_name(
                marker.display_name.as_deref(),
                merged_entry.display_name.as_deref(),
                repo_root.file_name().and_then(|value| value.to_str()),
            );
            let updated_marker =
                marker.rewrite_with_locality_id(target_locality_id, display_name)?;
            repo_marker::write(repo_root, &updated_marker).with_context(|| {
                format!(
                    "merge updated repo state and registry for locality_id `{target_locality_id}`, but failed to update the current clone marker"
                )
            })?;
            marker_updated = true;
        }
    }

    remove_registry_dir(&layout, source_locality_id).with_context(|| {
        format!(
            "merge updated repo state and registry for locality_id `{target_locality_id}`, but failed to remove the source registry locality_id `{source_locality_id}`"
        )
    })?;

    let mut warnings = Vec::new();
    let unresolved_clone_roots = source_entry
        .known_clone_roots
        .into_iter()
        .filter(|root| {
            if !marker_updated {
                return true;
            }
            match repo_registry::normalize_clone_root(repo_root) {
                Ok(current) => root != &current,
                Err(_) => true,
            }
        })
        .collect::<Vec<_>>();
    if !unresolved_clone_roots.is_empty() {
        warnings.push(format!(
            "other clone roots still need manual relink from locality_id `{source_locality_id}` to `{target_locality_id}`: {}",
            unresolved_clone_roots.join(", ")
        ));
    }
    if marker_updated {
        warnings.push(clone_state_warning(repo_root));
    }

    Ok(RepoMergeReport {
        command: "repo-merge",
        ok: true,
        path: repo_root.display().to_string(),
        profile: profile.to_string(),
        target_locality_id: target_locality_id.to_owned(),
        source_locality_id: source_locality_id.to_owned(),
        marker_updated,
        marker_path: marker_path.display().to_string(),
        registry_path: target_registry_path.display().to_string(),
        removed_registry_path: source_registry_path.display().to_string(),
        merged_overlay_paths: merged_state.overlay_paths,
        warnings,
    })
}

pub fn split(
    repo_root: &Path,
    explicit_profile: Option<&str>,
    source_locality_id: &str,
) -> Result<RepoSplitReport> {
    let profile = profile::resolve(explicit_profile)?;
    let layout = StateLayout::resolve(repo_root, profile.clone())?;
    let marker =
        repo_marker::load(repo_root)?.ok_or_else(|| linked_clone_error(repo_root, &layout))?;
    if marker.locality_id != source_locality_id {
        bail!(
            "current clone is linked to locality_id `{}`; only the active locality_id can be split in-place",
            marker.locality_id
        );
    }

    let source_registry_path = layout.repo_metadata_path(source_locality_id)?;
    let mut source_entry = repo_registry::load(&source_registry_path)?.ok_or_else(|| {
        anyhow::anyhow!(
            "locality_id `{source_locality_id}` is not registered at {}",
            source_registry_path.display()
        )
    })?;

    let current_clone_root = repo_registry::normalize_clone_root(repo_root)?;
    let aliases = link::collect_aliases(repo_root)?;
    let locality_id = link::generate_locality_id(repo_root, &layout)?;
    let display_name = link::choose_display_name(
        marker.display_name.as_deref(),
        source_entry.display_name.as_deref(),
        repo_root.file_name().and_then(|value| value.to_str()),
    );
    let registry_entry = RepoRegistryEntry::new_with_known_clone_roots(
        display_name.clone(),
        aliases,
        vec![current_clone_root.clone()],
    )?;

    let copied_state =
        copy_repo_state_across_profiles(&layout, source_locality_id, &locality_id).with_context(|| {
            format!(
                "split failed while copying repo state from locality_id `{source_locality_id}` into new locality_id `{locality_id}`; no registry changes were made"
            )
        })?;
    let registry_path = layout.repo_metadata_path(&locality_id)?;
    repo_registry::write(&registry_path, &registry_entry).with_context(|| {
        format!(
            "split copied repo state into new locality_id `{locality_id}`, but failed to write its registry at {}",
            registry_path.display()
        )
    })?;
    source_entry.remove_known_clone_root(&current_clone_root);
    repo_registry::write(&source_registry_path, &source_entry).with_context(|| {
        format!(
            "split wrote the new registry for locality_id `{locality_id}`, but failed to update the source registry locality_id `{source_locality_id}` at {}",
            source_registry_path.display()
        )
    })?;

    let updated_marker = marker.rewrite_with_locality_id(&locality_id, display_name)?;
    let marker_path = repo_marker::write(repo_root, &updated_marker).with_context(|| {
        format!(
            "split copied repo state and updated registries, but failed to update the current clone marker to locality_id `{locality_id}`"
        )
    })?;
    let overlay_path = layout.repo_overlay_root(&locality_id)?;

    let warnings = vec![clone_state_warning(repo_root)];

    Ok(RepoSplitReport {
        command: "repo-split",
        ok: true,
        path: repo_root.display().to_string(),
        profile: profile.to_string(),
        source_locality_id: source_locality_id.to_owned(),
        locality_id: locality_id.clone(),
        marker_path: marker_path.display().to_string(),
        registry_path: registry_path.display().to_string(),
        overlay_path: overlay_path.display().to_string(),
        copied_overlay_paths: copied_state.overlay_paths,
        warnings,
    })
}

fn load_registry_entries(
    layout: &StateLayout,
    current_locality_id: &Option<String>,
) -> Result<Vec<RepoListEntry>> {
    let registry_root = layout.repo_registry_root();
    if !registry_root.is_dir() {
        return Ok(Vec::new());
    }

    let mut repos = Vec::new();
    for entry in fs::read_dir(&registry_root)
        .with_context(|| format!("failed to read directory {}", registry_root.display()))?
    {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            continue;
        }

        let locality_id = entry.file_name().to_string_lossy().to_string();
        let registry_path = entry.path().join(repo_registry::REPO_METADATA_FILE);
        let Some(registry_entry) = repo_registry::load(&registry_path)? else {
            continue;
        };

        repos.push(build_repo_list_entry(
            layout,
            locality_id,
            registry_path,
            registry_entry,
            current_locality_id.as_deref(),
        )?);
    }

    repos.sort_by(|left, right| left.locality_id.cmp(&right.locality_id));
    Ok(repos)
}

fn build_repo_list_entry(
    layout: &StateLayout,
    locality_id: String,
    registry_path: PathBuf,
    registry_entry: RepoRegistryEntry,
    current_locality_id: Option<&str>,
) -> Result<RepoListEntry> {
    let overlay_path = layout.repo_overlay_root(&locality_id)?;

    Ok(RepoListEntry {
        current: current_locality_id == Some(locality_id.as_str()),
        locality_id,
        display_name: registry_entry.display_name,
        registry_path: registry_path.display().to_string(),
        overlay_path: overlay_path.display().to_string(),
        overlay_present: overlay_path.is_dir(),
        known_clone_roots: registry_entry.known_clone_roots,
        aliases: registry_entry.aliases,
    })
}

fn presence_label(present: bool) -> &'static str {
    if present {
        "present"
    } else {
        "missing"
    }
}

fn render_aliases(label: &str, aliases: &RepoAliases) {
    println!("{label}:");
    render_alias_group("remote_urls", &aliases.remote_urls);
    render_alias_group("root_commits", &aliases.root_commits);
    render_alias_group("repo_basenames", &aliases.repo_basenames);
}

fn render_alias_group(label: &str, values: &[String]) {
    if values.is_empty() {
        println!("  {label}: none");
        return;
    }

    println!("  {label}: {}", values.join(", "));
}

fn attach_command(repo_root: &Path, layout: &StateLayout) -> String {
    let base = format!("ccd attach --path {}", repo_root.display());
    if layout.profile().as_str() == DEFAULT_PROFILE {
        base
    } else {
        format!("{base} --profile {}", layout.profile())
    }
}

fn link_command(repo_root: &Path, layout: &StateLayout, locality_id: Option<&str>) -> String {
    let mut command = format!("ccd link --path {}", repo_root.display());
    if let Some(locality_id) = locality_id {
        command.push_str(&format!(" --project-id {locality_id}"));
    }
    if layout.profile().as_str() != DEFAULT_PROFILE {
        command.push_str(&format!(" --profile {}", layout.profile()));
    }
    command
}

fn linked_clone_error(repo_root: &Path, layout: &StateLayout) -> anyhow::Error {
    anyhow::anyhow!(
        "ccd repo cannot inspect or repair project linkage because this workspace is not linked: {} is missing. Run `{}` to bootstrap a new project overlay, or `{}` to reconnect to an existing one.",
        repo_root.join(repo_marker::MARKER_FILE).display(),
        attach_command(repo_root, layout),
        link_command(repo_root, layout, None)
    )
}

fn clone_state_warning(repo_root: &Path) -> String {
    format!(
        "workspace-local runtime state may still reflect the previous project linkage; run `ccd start --refresh --path {}` before relying on handoff or compiled state",
        repo_root.display()
    )
}

fn merge_registry_entries(
    target: RepoRegistryEntry,
    source: RepoRegistryEntry,
) -> Result<RepoRegistryEntry> {
    RepoRegistryEntry::new_with_known_clone_roots(
        target.display_name.or(source.display_name),
        RepoAliases {
            remote_urls: unique_strings(
                target
                    .aliases
                    .remote_urls
                    .into_iter()
                    .chain(source.aliases.remote_urls),
            ),
            root_commits: unique_strings(
                target
                    .aliases
                    .root_commits
                    .into_iter()
                    .chain(source.aliases.root_commits),
            ),
            repo_basenames: unique_strings(
                target
                    .aliases
                    .repo_basenames
                    .into_iter()
                    .chain(source.aliases.repo_basenames),
            ),
        },
        unique_strings(
            target
                .known_clone_roots
                .into_iter()
                .chain(source.known_clone_roots),
        ),
    )
}

fn collect_merge_conflicts(
    layout: &StateLayout,
    source_locality_id: &str,
    target_locality_id: &str,
) -> Result<Vec<String>> {
    let mut conflicts = Vec::new();
    for profile_layout in all_profile_layouts(layout)? {
        let source_overlay = profile_layout.repo_overlay_root(source_locality_id)?;
        let target_overlay = profile_layout.repo_overlay_root(target_locality_id)?;
        scan_tree_conflicts(
            &source_overlay,
            &source_overlay,
            &target_overlay,
            &mut conflicts,
        )?;
    }
    conflicts.sort();
    conflicts.dedup();
    Ok(conflicts)
}

fn merge_repo_state_across_profiles(
    layout: &StateLayout,
    source_locality_id: &str,
    target_locality_id: &str,
) -> Result<RepoStateChangeSummary> {
    let mut summary = RepoStateChangeSummary::default();

    for profile_layout in all_profile_layouts(layout)? {
        let source_overlay = profile_layout.repo_overlay_root(source_locality_id)?;
        let target_overlay = profile_layout.repo_overlay_root(target_locality_id)?;
        if merge_tree_into(&source_overlay, &source_overlay, &target_overlay)? {
            summary
                .overlay_paths
                .push(target_overlay.display().to_string());
        }
    }

    summary.overlay_paths = unique_strings(summary.overlay_paths);
    Ok(summary)
}

fn copy_repo_state_across_profiles(
    layout: &StateLayout,
    source_locality_id: &str,
    target_locality_id: &str,
) -> Result<RepoStateChangeSummary> {
    let mut summary = RepoStateChangeSummary::default();

    for profile_layout in all_profile_layouts(layout)? {
        let source_overlay = profile_layout.repo_overlay_root(source_locality_id)?;
        let target_overlay = profile_layout.repo_overlay_root(target_locality_id)?;
        if copy_tree(&source_overlay, &source_overlay, &target_overlay)? {
            summary
                .overlay_paths
                .push(target_overlay.display().to_string());
        }
    }

    summary.overlay_paths = unique_strings(summary.overlay_paths);
    Ok(summary)
}

fn all_profile_layouts(layout: &StateLayout) -> Result<Vec<StateLayout>> {
    let profiles_root = layout.ccd_root().join("profiles");
    let mut layouts = Vec::new();

    for profile_root in sorted_child_dirs(&profiles_root)? {
        let profile_name = profile_root
            .file_name()
            .map(|value| value.to_string_lossy().into_owned())
            .unwrap_or_default();
        let profile = ProfileName::new(profile_name)?;
        layouts.push(StateLayout::new_with_substrate(
            layout.ccd_root().to_path_buf(),
            layout.resolved_substrate().clone(),
            profile,
        ));
    }

    if layouts.is_empty() {
        layouts.push(layout.clone());
    }

    Ok(layouts)
}

fn scan_tree_conflicts(
    source_root: &Path,
    source: &Path,
    target: &Path,
    conflicts: &mut Vec<String>,
) -> Result<()> {
    match existing_path_kind(source)? {
        None => return Ok(()),
        Some(ExistingPathKind::Dir) => {}
        Some(ExistingPathKind::File) => {
            bail!(
                "unsupported repo state path at {}; expected a directory tree",
                source.display()
            )
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(source))
        }
    }

    match existing_path_kind(target)? {
        None | Some(ExistingPathKind::Dir) => {}
        Some(ExistingPathKind::File) => {
            conflicts.push(format!(
                "{} conflicts with existing file {}",
                source.display(),
                target.display()
            ));
            return Ok(());
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(target))
        }
    }

    for entry in fs::read_dir(source)
        .with_context(|| format!("failed to read directory {}", source.display()))?
    {
        let entry = entry
            .with_context(|| format!("failed to read directory entry in {}", source.display()))?;
        let source_path = entry.path();
        if !should_transfer_overlay_entry(source_root, &source_path)? {
            continue;
        }
        let target_path = target.join(entry.file_name());
        let file_type = entry.file_type().with_context(|| {
            format!("failed to inspect file type for {}", source_path.display())
        })?;
        ensure_supported_repo_state_entry(&source_path, &file_type)?;

        if file_type.is_dir() {
            match existing_path_kind(&target_path)? {
                Some(ExistingPathKind::File) => {
                    conflicts.push(format!(
                        "{} conflicts with existing file {}",
                        source_path.display(),
                        target_path.display()
                    ));
                    continue;
                }
                Some(ExistingPathKind::Unsupported) => {
                    bail!(unsupported_file_type_message(&target_path))
                }
                None | Some(ExistingPathKind::Dir) => {}
            }
            scan_tree_conflicts(source_root, &source_path, &target_path, conflicts)?;
            continue;
        }

        match existing_path_kind(&target_path)? {
            None => continue,
            Some(ExistingPathKind::Dir) => {
                conflicts.push(format!(
                    "{} conflicts with existing directory {}",
                    source_path.display(),
                    target_path.display()
                ));
                continue;
            }
            Some(ExistingPathKind::Unsupported) => {
                bail!(unsupported_file_type_message(&target_path))
            }
            Some(ExistingPathKind::File) => {
                if !file_contents_equal(&source_path, &target_path)? {
                    conflicts.push(format!(
                        "{} differs from {}",
                        source_path.display(),
                        target_path.display()
                    ));
                }
            }
        }
    }

    Ok(())
}

fn merge_tree_into(source_root: &Path, source: &Path, target: &Path) -> Result<bool> {
    match existing_path_kind(source)? {
        None => return Ok(false),
        Some(ExistingPathKind::Dir) => {}
        Some(ExistingPathKind::File) => {
            bail!(
                "cannot merge non-directory paths: {} -> {}",
                source.display(),
                target.display()
            )
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(source))
        }
    }

    match existing_path_kind(target)? {
        None => {
            fs::create_dir_all(target)
                .with_context(|| format!("failed to create {}", target.display()))?;
        }
        Some(ExistingPathKind::Dir) => {}
        Some(ExistingPathKind::File) => {
            bail!(
                "cannot merge non-directory paths: {} -> {}",
                source.display(),
                target.display()
            )
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(target))
        }
    }

    for entry in fs::read_dir(source)
        .with_context(|| format!("failed to read directory {}", source.display()))?
    {
        let entry = entry
            .with_context(|| format!("failed to read directory entry in {}", source.display()))?;
        let source_path = entry.path();
        if !should_transfer_overlay_entry(source_root, &source_path)? {
            continue;
        }
        let target_path = target.join(entry.file_name());
        let file_type = entry.file_type().with_context(|| {
            format!("failed to inspect file type for {}", source_path.display())
        })?;
        ensure_supported_repo_state_entry(&source_path, &file_type)?;

        if file_type.is_dir() {
            merge_tree_into(source_root, &source_path, &target_path)?;
        } else if file_type.is_file() {
            merge_file_into(&source_path, &target_path)?;
        }
    }

    fs::remove_dir_all(source).with_context(|| format!("failed to remove {}", source.display()))?;
    Ok(true)
}

fn merge_file_into(source: &Path, target: &Path) -> Result<bool> {
    match existing_path_kind(source)? {
        None => return Ok(false),
        Some(ExistingPathKind::File) => {}
        Some(ExistingPathKind::Dir) => {
            bail!(
                "cannot merge directory {} into file {}",
                source.display(),
                target.display()
            )
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(source))
        }
    }

    match existing_path_kind(target)? {
        None => {
            move_file_into_empty_target(source, target)?;
            return Ok(true);
        }
        Some(ExistingPathKind::Dir) => {
            bail!(
                "cannot merge file {} into directory {}",
                source.display(),
                target.display()
            )
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(target))
        }
        Some(ExistingPathKind::File) => {}
    }

    if !file_contents_equal(source, target)? {
        bail!(
            "refusing to overwrite differing file {} with {}",
            target.display(),
            source.display()
        );
    }
    fs::remove_file(source).with_context(|| format!("failed to remove {}", source.display()))?;
    Ok(true)
}

fn copy_tree(source_root: &Path, source: &Path, target: &Path) -> Result<bool> {
    match existing_path_kind(source)? {
        None => return Ok(false),
        Some(ExistingPathKind::File) => return copy_file(source, target),
        Some(ExistingPathKind::Dir) => {}
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(source))
        }
    }

    if existing_path_kind(target)?.is_some() {
        bail!(
            "refusing to copy {} into existing path {}",
            source.display(),
            target.display()
        );
    }

    fs::create_dir_all(target).with_context(|| format!("failed to create {}", target.display()))?;

    for entry in fs::read_dir(source)
        .with_context(|| format!("failed to read directory {}", source.display()))?
    {
        let entry = entry
            .with_context(|| format!("failed to read directory entry in {}", source.display()))?;
        let source_path = entry.path();
        if !should_transfer_overlay_entry(source_root, &source_path)? {
            continue;
        }
        let target_path = target.join(entry.file_name());
        let file_type = entry.file_type().with_context(|| {
            format!("failed to inspect file type for {}", source_path.display())
        })?;
        ensure_supported_repo_state_entry(&source_path, &file_type)?;

        if file_type.is_dir() {
            copy_tree(source_root, &source_path, &target_path)?;
        } else if file_type.is_file() {
            copy_file(&source_path, &target_path)?;
        }
    }

    Ok(true)
}

fn copy_file(source: &Path, target: &Path) -> Result<bool> {
    match existing_path_kind(source)? {
        None => return Ok(false),
        Some(ExistingPathKind::File) => {}
        Some(ExistingPathKind::Dir) => {
            bail!(
                "cannot copy directory {} into file path {}",
                source.display(),
                target.display()
            )
        }
        Some(ExistingPathKind::Unsupported) => {
            bail!(unsupported_file_type_message(source))
        }
    }

    if existing_path_kind(target)?.is_some() {
        bail!(
            "refusing to copy {} into existing path {}",
            source.display(),
            target.display()
        );
    }

    create_parent_dir(target)?;
    fs::copy(source, target).with_context(|| {
        format!(
            "failed to copy {} to {}",
            source.display(),
            target.display()
        )
    })?;
    Ok(true)
}

fn should_transfer_overlay_entry(source_root: &Path, source_path: &Path) -> Result<bool> {
    let relative_path = source_path.strip_prefix(source_root).with_context(|| {
        format!(
            "failed to compute repo overlay relative path for {} under {}",
            source_path.display(),
            source_root.display()
        )
    })?;
    let under_extensions = matches!(
        relative_path.components().next(),
        Some(Component::Normal(name)) if name == OsStr::new("extensions")
    );
    Ok(!under_extensions)
}

fn file_contents_equal(left: &Path, right: &Path) -> Result<bool> {
    let left_len = fs::metadata(left)
        .with_context(|| format!("failed to read metadata for {}", left.display()))?
        .len();
    let right_len = fs::metadata(right)
        .with_context(|| format!("failed to read metadata for {}", right.display()))?
        .len();
    if left_len != right_len {
        return Ok(false);
    }

    let left_bytes =
        fs::read(left).with_context(|| format!("failed to read {}", left.display()))?;
    let right_bytes =
        fs::read(right).with_context(|| format!("failed to read {}", right.display()))?;
    Ok(left_bytes == right_bytes)
}

fn create_parent_dir(path: &Path) -> Result<()> {
    let parent = path
        .parent()
        .ok_or_else(|| anyhow::anyhow!("path has no parent: {}", path.display()))?;
    fs::create_dir_all(parent)
        .with_context(|| format!("failed to create directory {}", parent.display()))
}

fn remove_registry_dir(layout: &StateLayout, locality_id: &str) -> Result<()> {
    let registry_dir = layout.repo_registry_dir(locality_id)?;
    if registry_dir.is_dir() {
        fs::remove_dir_all(&registry_dir)
            .with_context(|| format!("failed to remove {}", registry_dir.display()))?;
    }
    Ok(())
}

fn unique_strings<I>(values: I) -> Vec<String>
where
    I: IntoIterator<Item = String>,
{
    let mut seen = BTreeSet::new();
    let mut unique = Vec::new();
    for value in values {
        if seen.insert(value.clone()) {
            unique.push(value);
        }
    }
    unique
}

fn sorted_child_dirs(root: &Path) -> Result<Vec<PathBuf>> {
    let read_dir = match fs::read_dir(root) {
        Ok(read_dir) => read_dir,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(error) => {
            return Err(error)
                .with_context(|| format!("failed to read directory {}", root.display()))
        }
    };

    let mut entries = Vec::new();
    for entry in read_dir {
        let entry = entry
            .with_context(|| format!("failed to read directory entry in {}", root.display()))?;
        let path = entry.path();
        let file_type = entry
            .file_type()
            .with_context(|| format!("failed to inspect file type for {}", path.display()))?;
        if file_type.is_dir() {
            entries.push(path);
        }
    }
    entries.sort();
    Ok(entries)
}

fn existing_path_kind(path: &Path) -> Result<Option<ExistingPathKind>> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => {
            let file_type = metadata.file_type();
            if file_type.is_dir() {
                Ok(Some(ExistingPathKind::Dir))
            } else if file_type.is_file() {
                Ok(Some(ExistingPathKind::File))
            } else {
                Ok(Some(ExistingPathKind::Unsupported))
            }
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(error).with_context(|| format!("failed to stat {}", path.display())),
    }
}

fn ensure_supported_repo_state_entry(path: &Path, file_type: &fs::FileType) -> Result<()> {
    if file_type.is_dir() || file_type.is_file() {
        return Ok(());
    }

    bail!(unsupported_file_type_message(path))
}

fn unsupported_file_type_message(path: &Path) -> String {
    format!(
        "unsupported file type at {}; only regular files and directories can be merged or copied",
        path.display()
    )
}

fn move_file_into_empty_target(source: &Path, target: &Path) -> Result<()> {
    create_parent_dir(target)?;
    match fs::rename(source, target) {
        Ok(()) => Ok(()),
        Err(error) if is_cross_device_rename(&error) => {
            copy_file(source, target).with_context(|| {
                format!(
                    "failed to copy {} to {} after cross-filesystem move fallback",
                    source.display(),
                    target.display()
                )
            })?;
            fs::remove_file(source).with_context(|| {
                format!(
                    "failed to remove {} after cross-filesystem copy to {}",
                    source.display(),
                    target.display()
                )
            })?;
            Ok(())
        }
        Err(error) => Err(error).with_context(|| {
            format!(
                "failed to move {} to {}",
                source.display(),
                target.display()
            )
        }),
    }
}

const EXDEV_OS_ERROR: i32 = 18;

fn is_cross_device_rename(error: &std::io::Error) -> bool {
    error.raw_os_error() == Some(EXDEV_OS_ERROR)
}
