//! Reconcile divergent task histories on `tsk git-pull`.
//!
//! `git_pull` fetches the remote's `refs/tsk/*` into a non-clobbering shadow
//! namespace at `refs/tsk-fetched/<remote>/*` so both the local and remote
//! tip of every task ref are available in the same repo. We then walk every
//! task that exists in either, and for each:
//!
//! - one side missing → take the side that has it
//! - one side strictly ancestor of the other → fast-forward (or no-op)
//! - both diverged → reconcile per [`Strategy`]
//!
//! `Strategy::Merge` (default) creates a merge commit using `git2`'s
//! 3-way `merge_trees` against the common ancestor; clean merges land as a
//! single commit with two parents and the local user as both author and
//! committer. `Strategy::Rebase` replays each local-only commit on top of
//! the remote tip, preserving each commit's original author and updating
//! the committer to the local user — same shape as `git rebase`.
//!
//! True content conflicts (both sides edited the same blob in incompatible
//! ways) are committed with conflict markers in the task blob. That keeps the
//! two-parent merge history auditable while letting the user resolve with the
//! normal task/property editing commands.

use crate::errors::Result;
use crate::namespace::{self, NS_REF_PREFIX, Namespace};
use crate::object::{self, StableId, TASK_REF_PREFIX};
use crate::properties;
use crate::queue::{self, QUEUE_REF_PREFIX, Queue};
use crate::references;
use git2::{Commit, IndexEntry, Oid, Repository, Tree};
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::ffi::OsStr;
use std::path::Path;
use std::process::{Command, ExitStatus};

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum Strategy {
    #[default]
    Merge,
    Rebase,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReconKind {
    /// No work needed (either side strict ancestor or refs identical).
    Unchanged,
    /// Local was strict ancestor of remote; ref now points at remote tip.
    FastForward,
    /// Remote ref existed without a local counterpart; copied verbatim.
    NewRemote,
    /// Wrote a merge commit with two parents.
    Merged,
    /// Replayed local-only commits onto the remote tip.
    Rebased,
    /// Wrote a merge commit containing conflict markers for overlapping edits.
    Conflict,
}

#[derive(Debug)]
pub struct Reconciliation {
    pub stable: StableId,
    pub kind: ReconKind,
}

#[derive(Debug)]
pub struct PullOutcome {
    pub tasks: Vec<Reconciliation>,
    pub namespaces: Vec<NamespaceReconciliation>,
    pub queues: Vec<QueueReconciliation>,
}

pub const FETCH_PREFIX: &str = "refs/tsk-fetched/";

pub fn fetched_prefix(remote: &str) -> String {
    format!("{FETCH_PREFIX}{remote}/")
}

/// Reconcile every `refs/tsk/tasks/*` against its fetched counterpart at
/// `refs/tsk-fetched/<remote>/tasks/*`. Returns one entry per task that
/// existed in either side.
#[allow(dead_code)] // kept for callers that do not have a workspace namespace.
pub fn reconcile_task_refs(
    repo: &Repository,
    remote: &str,
    strategy: Strategy,
) -> Result<Vec<Reconciliation>> {
    reconcile_task_refs_in_namespace(repo, remote, namespace::DEFAULT_NS, strategy)
}

pub fn reconcile_task_refs_in_namespace(
    repo: &Repository,
    remote: &str,
    active_namespace: &str,
    strategy: Strategy,
) -> Result<Vec<Reconciliation>> {
    let fetched_tasks = format!("{}tasks/", fetched_prefix(remote));
    let mut stables: BTreeSet<String> = BTreeSet::new();
    for r in repo.references_glob(&format!("{TASK_REF_PREFIX}*"))? {
        let r = r?;
        if let Ok(name) = r.name()
            && let Some(name) = name.strip_prefix(TASK_REF_PREFIX)
        {
            stables.insert(name.to_string());
        }
    }
    for r in repo.references_glob(&format!("{fetched_tasks}*"))? {
        let r = r?;
        if let Ok(name) = r.name()
            && let Some(name) = name.strip_prefix(fetched_tasks.as_str())
        {
            stables.insert(name.to_string());
        }
    }
    let mut out = Vec::new();
    for s in stables {
        let stable = StableId(s.clone());
        let local = repo
            .find_reference(&stable.refname())
            .ok()
            .and_then(|r| r.target());
        let remote_ref = format!("{fetched_tasks}{s}");
        let remote_tip = repo
            .find_reference(&remote_ref)
            .ok()
            .and_then(|r| r.target());
        let kind = reconcile_one(repo, &stable, local, remote_tip, active_namespace, strategy)?;
        out.push(Reconciliation { stable, kind });
    }
    Ok(out)
}

fn reconcile_one(
    repo: &Repository,
    stable: &StableId,
    local: Option<Oid>,
    remote: Option<Oid>,
    active_namespace: &str,
    strategy: Strategy,
) -> Result<ReconKind> {
    match (local, remote) {
        (None, None) | (Some(_), None) => Ok(ReconKind::Unchanged),
        (None, Some(r)) => {
            repo.reference(&stable.refname(), r, true, "pull-import")?;
            Ok(ReconKind::NewRemote)
        }
        (Some(l), Some(r)) if l == r => Ok(ReconKind::Unchanged),
        (Some(l), Some(r)) => {
            // graph_descendant_of(a, b) is true iff a descends from b.
            if repo.graph_descendant_of(l, r).unwrap_or(false) {
                Ok(ReconKind::Unchanged)
            } else if repo.graph_descendant_of(r, l).unwrap_or(false) {
                repo.reference(&stable.refname(), r, true, "fast-forward")?;
                Ok(ReconKind::FastForward)
            } else {
                match strategy {
                    Strategy::Merge => {
                        merge_strategy(repo, stable, l, r, active_namespace, edit_conflict_content)
                    }
                    Strategy::Rebase => rebase_strategy(repo, stable, l, r),
                }
            }
        }
    }
}

fn merge_strategy(
    repo: &Repository,
    stable: &StableId,
    local: Oid,
    remote: Oid,
    active_namespace: &str,
    edit_conflict: impl FnMut(String) -> Result<Option<String>>,
) -> Result<ReconKind> {
    merge_strategy_inner(repo, stable, local, remote, active_namespace, edit_conflict)
}

fn merge_strategy_inner(
    repo: &Repository,
    stable: &StableId,
    local: Oid,
    remote: Oid,
    active_namespace: &str,
    mut edit_conflict: impl FnMut(String) -> Result<Option<String>>,
) -> Result<ReconKind> {
    let base_oid = repo.merge_base(local, remote)?;
    let base_tree = repo.find_commit(base_oid)?.tree()?;
    let local_commit = repo.find_commit(local)?;
    let remote_commit = repo.find_commit(remote)?;
    let our_tree = local_commit.tree()?;
    let their_tree = remote_commit.tree()?;
    let mut idx = repo.merge_trees(&base_tree, &our_tree, &their_tree, None)?;
    if idx.has_conflicts() {
        let resolution = write_conflict_resolution_tree(
            repo,
            &base_tree,
            &our_tree,
            &their_tree,
            &mut idx,
            local_commit.time().seconds(),
            remote_commit.time().seconds(),
            &mut edit_conflict,
        )?;
        let Some(tree_oid) = resolution.tree_oid else {
            return Ok(ReconKind::Conflict);
        };
        let sig = object::signature(repo);
        let parents: Vec<&Commit> = vec![&local_commit, &remote_commit];
        let short = &stable.0[..12.min(stable.0.len())];
        let mut message = format!("merge-conflict tsk-{short}");
        if !resolution.property_logs.is_empty() {
            message.push_str("\n\n");
            message.push_str(&resolution.property_logs.join("\n"));
        }
        let merge_oid = repo.commit(
            None,
            &sig,
            &sig,
            &message,
            &repo.find_tree(tree_oid)?,
            &parents,
        )?;
        repo.reference(&stable.refname(), merge_oid, true, "merge-conflict")?;
        refresh_merged_task_references(repo, stable, active_namespace)?;
        return Ok(ReconKind::Conflict);
    }
    let tree_oid = idx.write_tree_to(repo)?;
    let sig = object::signature(repo);
    let local_commit = repo.find_commit(local)?;
    let remote_commit = repo.find_commit(remote)?;
    let parents: Vec<&Commit> = vec![&local_commit, &remote_commit];
    let short = &stable.0[..12.min(stable.0.len())];
    let merge_oid = repo.commit(
        None,
        &sig,
        &sig,
        &format!("merge tsk-{short}"),
        &repo.find_tree(tree_oid)?,
        &parents,
    )?;
    repo.reference(&stable.refname(), merge_oid, true, "merge")?;
    refresh_merged_task_references(repo, stable, active_namespace)?;
    Ok(ReconKind::Merged)
}

fn edit_conflict_content(initial: String) -> Result<Option<String>> {
    let editor = std::env::var_os("VISUAL")
        .or_else(|| std::env::var_os("EDITOR"))
        .ok_or_else(|| {
            crate::errors::Error::Parse(
                "task body conflict requires VISUAL or EDITOR to resolve".into(),
            )
        })?;
    let path = conflict_temp_path()?;
    std::fs::write(&path, initial)?;
    let status = run_editor_command(&editor, &path)?;
    if !status.success() {
        return Err(crate::errors::Error::Parse(
            "editor exited unsuccessfully while resolving task body conflict".into(),
        ));
    }
    let edited = std::fs::read_to_string(&path)?;
    let _ = std::fs::remove_file(&path);
    if edited.trim().is_empty() {
        Ok(None)
    } else {
        Ok(Some(edited))
    }
}

fn run_editor_command(editor: &OsStr, path: &Path) -> Result<ExitStatus> {
    let command = editor.to_string_lossy();
    if command.trim().is_empty() {
        return Err(crate::errors::Error::Parse(
            "task body conflict requires VISUAL or EDITOR to resolve".into(),
        ));
    }

    #[cfg(windows)]
    {
        Ok(Command::new("cmd")
            .arg("/C")
            .arg(format!("{command} \"{}\"", path.display()))
            .status()?)
    }

    #[cfg(not(windows))]
    {
        Ok(Command::new("sh")
            .arg("-c")
            .arg(format!("{command} \"$1\""))
            .arg("tsk-editor")
            .arg(path)
            .status()?)
    }
}

fn conflict_temp_path() -> Result<std::path::PathBuf> {
    let mut path = std::env::temp_dir();
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_nanos();
    path.push(format!("tsk-conflict-{}-{nanos}.tmp", std::process::id()));
    Ok(path)
}

fn refresh_merged_task_references(
    repo: &Repository,
    stable: &StableId,
    active_namespace: &str,
) -> Result<()> {
    let Some(mut task) = object::read(repo, stable)? else {
        return Ok(());
    };
    let bindings = references::binding_map(repo, |s| object::exists(repo, s))?;
    let Some((source_namespace, source_id)) =
        references::canonical_binding(&bindings, active_namespace, stable)
    else {
        properties::reindex_task(repo, stable, &task.properties)?;
        return Ok(());
    };
    let refresh = references::refresh_task(repo, &source_namespace, &mut task, source_id, stable)?;
    properties::update_task(repo, stable, &task, "refresh-references")?;
    references::sync_referenced_by(
        repo,
        stable,
        &refresh.source_ref,
        &refresh.old_refs,
        &refresh.new_refs,
    )?;
    Ok(())
}

struct ConflictResolution {
    tree_oid: Option<Oid>,
    property_logs: Vec<String>,
}

fn write_conflict_resolution_tree(
    repo: &Repository,
    base_tree: &Tree<'_>,
    our_tree: &Tree<'_>,
    their_tree: &Tree<'_>,
    idx: &mut git2::Index,
    local_time: i64,
    remote_time: i64,
    edit_conflict: &mut impl FnMut(String) -> Result<Option<String>>,
) -> Result<ConflictResolution> {
    let mut tb = repo.treebuilder(None)?;
    for entry in idx.iter().filter(|entry| index_stage(entry) == 0) {
        tb.insert(
            path_str(&entry)?,
            entry.id,
            i32::try_from(entry.mode).unwrap_or(0o100644),
        )?;
    }

    let mut property_logs = Vec::new();
    let conflicts = idx.conflicts()?;
    for conflict in conflicts {
        let conflict = conflict?;
        let path = conflict
            .our
            .as_ref()
            .or(conflict.their.as_ref())
            .or(conflict.ancestor.as_ref())
            .ok_or_else(|| crate::errors::Error::Parse("empty merge conflict entry".into()))?;
        let path = path_str(path)?.to_string();
        let ours = conflict_blob(repo, our_tree, conflict.our.as_ref(), &path)?;
        let base = conflict_blob(repo, base_tree, conflict.ancestor.as_ref(), &path)?;
        let theirs = conflict_blob(repo, their_tree, conflict.their.as_ref(), &path)?;
        if path == object::CONTENT_FILE {
            let ours = String::from_utf8_lossy(ours.as_deref().unwrap_or_default());
            let base = String::from_utf8_lossy(base.as_deref().unwrap_or_default());
            let theirs = String::from_utf8_lossy(theirs.as_deref().unwrap_or_default());
            let initial = conflict_markers(&ours, Some(&base), &theirs);
            let Some(edited) = edit_conflict(initial)? else {
                return Ok(ConflictResolution {
                    tree_oid: None,
                    property_logs,
                });
            };
            if edited.trim().is_empty() {
                return Ok(ConflictResolution {
                    tree_oid: None,
                    property_logs,
                });
            };
            let bytes = edited.into_bytes();
            let oid = repo.blob(&bytes)?;
            tb.insert(path.as_str(), oid, 0o100644)?;
        } else {
            let (chosen, chosen_side) = if local_time > remote_time {
                (ours.as_deref(), "ours")
            } else {
                (theirs.as_deref(), "theirs")
            };
            if let Some(bytes) = chosen {
                let oid = repo.blob(bytes)?;
                tb.insert(path.as_str(), oid, 0o100644)?;
            }
            property_logs.push(format!(
                "property conflict {path}: chose {chosen_side} by timestamp \
                 (ours={local_time}, theirs={remote_time})"
            ));
        }
    }

    Ok(ConflictResolution {
        tree_oid: Some(tb.write()?),
        property_logs,
    })
}

fn path_str(entry: &IndexEntry) -> Result<&str> {
    std::str::from_utf8(&entry.path)
        .map_err(|err| crate::errors::Error::Parse(format!("invalid task tree path: {err}")))
}

fn index_stage(entry: &IndexEntry) -> u16 {
    (entry.flags >> 12) & 0x3
}

fn conflict_blob(
    repo: &Repository,
    tree: &Tree<'_>,
    entry: Option<&IndexEntry>,
    path: &str,
) -> Result<Option<Vec<u8>>> {
    if entry.is_none() {
        return Ok(None);
    }
    let Ok(entry) = tree.get_path(std::path::Path::new(path)) else {
        return Ok(None);
    };
    let blob = entry.to_object(repo)?.peel_to_blob()?;
    Ok(Some(blob.content().to_vec()))
}

fn conflict_markers(ours: &str, base: Option<&str>, theirs: &str) -> String {
    let mut out = String::new();
    out.push_str("<<<<<<< ours\n");
    out.push_str(ours);
    ensure_trailing_newline(&mut out);
    if let Some(base) = base {
        out.push_str("||||||| base\n");
        out.push_str(base);
        ensure_trailing_newline(&mut out);
    }
    out.push_str("=======\n");
    out.push_str(theirs);
    ensure_trailing_newline(&mut out);
    out.push_str(">>>>>>> theirs\n");
    out
}

fn ensure_trailing_newline(s: &mut String) {
    if !s.ends_with('\n') {
        s.push('\n');
    }
}

fn rebase_strategy(
    repo: &Repository,
    stable: &StableId,
    local: Oid,
    remote: Oid,
) -> Result<ReconKind> {
    let base_oid = repo.merge_base(local, remote)?;
    // Walk local from tip back to (but not including) base, then reverse so
    // we replay oldest-first.
    let mut to_replay: Vec<Oid> = Vec::new();
    let mut cur = repo.find_commit(local)?;
    while cur.id() != base_oid {
        to_replay.push(cur.id());
        let Ok(parent) = cur.parent(0) else { break };
        cur = parent;
    }
    to_replay.reverse();
    let committer = object::signature(repo);
    let mut current = remote;
    for c_oid in to_replay {
        let c = repo.find_commit(c_oid)?;
        let parent_tree = c.parent(0)?.tree()?;
        let c_tree = c.tree()?;
        let cur_commit = repo.find_commit(current)?;
        let cur_tree = cur_commit.tree()?;
        let mut idx = repo.merge_trees(&parent_tree, &cur_tree, &c_tree, None)?;
        if idx.has_conflicts() {
            return Ok(ReconKind::Conflict);
        }
        let tree_oid = idx.write_tree_to(repo)?;
        let new_oid = repo.commit(
            None,
            &c.author(),
            &committer,
            c.message().unwrap_or(""),
            &repo.find_tree(tree_oid)?,
            &[&cur_commit],
        )?;
        current = new_oid;
    }
    repo.reference(&stable.refname(), current, true, "rebase")?;
    Ok(ReconKind::Rebased)
}

/// One namespace's reconciliation outcome at pull time.
#[derive(Debug)]
pub struct NamespaceReconciliation {
    pub namespace: String,
    /// `(old_human_id, new_human_id)` per binding that had to be moved
    /// because the remote claimed the same id for a different stable.
    pub renumbers: Vec<(u32, u32)>,
}

/// Three-way merge each namespace ref against its fetched counterpart.
///
/// On conflict (same human id mapped to different stable ids on each
/// side), the **remote** binding keeps the id and the **local** binding
/// is renumbered to a fresh id past `max(local.next, remote.next)`.
/// Local-only bindings are preserved at their original id; remote-only
/// bindings are added verbatim. The merged tree is written as a commit
/// with two parents so future pulls can fast-forward.
pub fn reconcile_namespace_refs(
    repo: &Repository,
    remote: &str,
) -> Result<Vec<NamespaceReconciliation>> {
    let fetched_ns = format!("{}namespaces/", fetched_prefix(remote));
    let mut names: BTreeSet<String> = BTreeSet::new();
    for r in repo.references_glob(&format!("{NS_REF_PREFIX}*"))? {
        let r = r?;
        if let Ok(name) = r.name()
            && let Some(name) = name.strip_prefix(NS_REF_PREFIX)
        {
            names.insert(name.to_string());
        }
    }
    for r in repo.references_glob(&format!("{fetched_ns}*"))? {
        let r = r?;
        if let Ok(name) = r.name()
            && let Some(name) = name.strip_prefix(fetched_ns.as_str())
        {
            names.insert(name.to_string());
        }
    }
    let mut out = Vec::new();
    for name in names {
        let local = repo
            .find_reference(&namespace::refname(&name))
            .ok()
            .and_then(|r| r.target());
        let remote_oid = repo
            .find_reference(&format!("{fetched_ns}{name}"))
            .ok()
            .and_then(|r| r.target());
        if let Some(rec) = reconcile_namespace_one(repo, &name, local, remote_oid)? {
            out.push(rec);
        }
    }
    Ok(out)
}

fn reconcile_namespace_one(
    repo: &Repository,
    name: &str,
    local: Option<Oid>,
    remote: Option<Oid>,
) -> Result<Option<NamespaceReconciliation>> {
    match (local, remote) {
        (None, None) | (Some(_), None) => Ok(None),
        (None, Some(r)) => {
            repo.reference(&namespace::refname(name), r, true, "pull-import")?;
            Ok(None)
        }
        (Some(l), Some(r)) if l == r => Ok(None),
        (Some(l), Some(r)) => {
            if repo.graph_descendant_of(l, r).unwrap_or(false) {
                return Ok(None);
            }
            if repo.graph_descendant_of(r, l).unwrap_or(false) {
                repo.reference(&namespace::refname(name), r, true, "fast-forward")?;
                return Ok(None);
            }
            // Diverged: 3-way merge with remote-wins on conflicts.
            let local_ns = namespace::read_at_commit(repo, l)?;
            let remote_ns = namespace::read_at_commit(repo, r)?;
            let mut merged = Namespace {
                next: local_ns.next.max(remote_ns.next),
                mapping: remote_ns.mapping.clone(),
            };
            let mut renumbers: Vec<(u32, u32)> = Vec::new();
            for (lh, lstable) in &local_ns.mapping {
                match remote_ns.mapping.get(lh) {
                    Some(rstable) if rstable == lstable => {} // already in merged
                    Some(_rstable) => {
                        // Conflict: same id, different stable. Renumber local.
                        let new_h = merged.next;
                        merged.next += 1;
                        merged.mapping.insert(new_h, lstable.clone());
                        renumbers.push((*lh, new_h));
                    }
                    None => {
                        // Local-only binding; preserve at its current id (it
                        // can't collide because remote_ns lacks that id).
                        merged.mapping.insert(*lh, lstable.clone());
                    }
                }
            }
            // Bump next past any human id we just placed.
            if let Some(max_h) = merged.mapping.keys().max() {
                merged.next = merged.next.max(max_h + 1);
            }
            // Write a merge commit with two parents.
            let tree_oid = namespace::build_tree(repo, &merged)?;
            let local_commit = repo.find_commit(l)?;
            let remote_commit = repo.find_commit(r)?;
            let sig = object::signature(repo);
            let msg = if renumbers.is_empty() {
                format!("merge-namespace {name}")
            } else {
                format!("rebase-bind {name}")
            };
            let parents: Vec<&Commit> = vec![&local_commit, &remote_commit];
            let new_oid =
                repo.commit(None, &sig, &sig, &msg, &repo.find_tree(tree_oid)?, &parents)?;
            repo.reference(&namespace::refname(name), new_oid, true, &msg)?;
            if !renumbers.is_empty() {
                adjust_renumbered_internal_links(repo, &local_ns, &renumbers)?;
            }
            Ok(Some(NamespaceReconciliation {
                namespace: name.to_string(),
                renumbers,
            }))
        }
    }
}

fn adjust_renumbered_internal_links(
    repo: &Repository,
    local_ns: &Namespace,
    renumbers: &[(u32, u32)],
) -> Result<()> {
    let mut seen = HashSet::new();
    for stable in local_ns.mapping.values() {
        if !seen.insert(stable.clone()) {
            continue;
        }
        let Some(mut task) = object::read(repo, stable)? else {
            continue;
        };
        let original = task.content.clone();
        for (old, new) in renumbers {
            task.content = task
                .content
                .replace(&format!("[[tsk-{old}]]"), &format!("[[tsk-{new}]]"));
        }
        if task.content != original {
            properties::update_task(repo, stable, &task, "rewrite-renumbered-links")?;
        }
    }

    for stable in object::list_all(repo)? {
        let Some(mut task) = object::read(repo, &stable)? else {
            continue;
        };
        if references::rewrite_renumbered_properties(&mut task.properties, renumbers) {
            properties::update_task(
                repo,
                &stable,
                &task,
                "rewrite-renumbered-reference-properties",
            )?;
        }
    }
    Ok(())
}

/// One queue's reconciliation outcome at pull time. Currently we only
/// surface that a non-trivial merge happened; counts/diffs could be
/// added later if useful.
#[derive(Debug)]
pub struct QueueReconciliation {
    pub name: String,
    pub kind: QueueReconKind,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum QueueReconKind {
    Merged,
    Deleted,
}

/// 3-way merge each queue ref against its fetched counterpart.
///
/// `index`: per-stable-id 3-way set merge — entries present in *base*
/// stay only if both sides keep them; entries added on either side are
/// included; entries removed on either side are dropped. Order is
/// remote-first, then local additions appended.
///
/// `inbox`: per-key 3-way map merge. Removals on either side win;
/// adds on either side are included; if both sides set the same new key
/// to different stables (shouldn't happen — keys carry a per-source
/// sequence), remote wins.
///
/// `can_pull`: 3-way bool. Local change wins if it differs from base;
/// otherwise take remote.
pub fn reconcile_queue_refs_with_deletions(
    repo: &Repository,
    remote: &str,
    remotely_deleted: &BTreeSet<String>,
) -> Result<Vec<QueueReconciliation>> {
    let fetched = format!("{}queues/", fetched_prefix(remote));
    let mut names: BTreeSet<String> = BTreeSet::new();
    for r in repo.references_glob(&format!("{QUEUE_REF_PREFIX}*"))? {
        let r = r?;
        if let Ok(n) = r.name()
            && let Some(n) = n.strip_prefix(QUEUE_REF_PREFIX)
        {
            names.insert(n.to_string());
        }
    }
    for r in repo.references_glob(&format!("{fetched}*"))? {
        let r = r?;
        if let Ok(n) = r.name()
            && let Some(n) = n.strip_prefix(fetched.as_str())
        {
            names.insert(n.to_string());
        }
    }
    let mut out = Vec::new();
    for name in names {
        let local = repo
            .find_reference(&queue::refname(&name))
            .ok()
            .and_then(|r| r.target());
        let remote_oid = repo
            .find_reference(&format!("{fetched}{name}"))
            .ok()
            .and_then(|r| r.target());
        if remote_oid.is_none()
            && local.is_some()
            && remotely_deleted.contains(&name)
            && queue::delete(repo, &name)?
        {
            out.push(QueueReconciliation {
                name,
                kind: QueueReconKind::Deleted,
            });
            continue;
        }
        if let Some(rec) = reconcile_queue_one(repo, &name, local, remote_oid)? {
            out.push(rec);
        }
    }
    Ok(out)
}

fn reconcile_queue_one(
    repo: &Repository,
    name: &str,
    local: Option<Oid>,
    remote: Option<Oid>,
) -> Result<Option<QueueReconciliation>> {
    match (local, remote) {
        (None, None) | (Some(_), None) => Ok(None),
        (None, Some(r)) => {
            repo.reference(&queue::refname(name), r, true, "pull-import")?;
            Ok(None)
        }
        (Some(l), Some(r)) if l == r => Ok(None),
        (Some(l), Some(r)) => {
            if repo.graph_descendant_of(l, r).unwrap_or(false) {
                return Ok(None);
            }
            if repo.graph_descendant_of(r, l).unwrap_or(false) {
                repo.reference(&queue::refname(name), r, true, "fast-forward")?;
                return Ok(None);
            }
            // No common ancestor when each clone independently rooted
            // its queue ref; treat the base as empty in that case.
            let base_q = match repo.merge_base(l, r).ok() {
                Some(oid) => queue::read_at_commit(repo, name, oid)?,
                None => Queue::new(name),
            };
            let local_q = queue::read_at_commit(repo, name, l)?;
            let remote_q = queue::read_at_commit(repo, name, r)?;
            let merged = three_way_queue_merge(&base_q, &local_q, &remote_q);
            let tree_oid = queue::build_tree(repo, &merged)?;
            let sig = object::signature(repo);
            let local_commit = repo.find_commit(l)?;
            let remote_commit = repo.find_commit(r)?;
            let parents: Vec<&Commit> = vec![&local_commit, &remote_commit];
            let new_oid = repo.commit(
                None,
                &sig,
                &sig,
                &format!("merge-queue {name}"),
                &repo.find_tree(tree_oid)?,
                &parents,
            )?;
            repo.reference(&queue::refname(name), new_oid, true, "merge")?;
            Ok(Some(QueueReconciliation {
                name: name.to_string(),
                kind: QueueReconKind::Merged,
            }))
        }
    }
}

fn three_way_queue_merge(base: &Queue, local: &Queue, remote: &Queue) -> Queue {
    let base_set: HashSet<&StableId> = base.index.iter().collect();
    let local_set: HashSet<&StableId> = local.index.iter().collect();
    let remote_set: HashSet<&StableId> = remote.index.iter().collect();
    let keep: HashSet<&StableId> = base_set
        .iter()
        .chain(local_set.iter())
        .chain(remote_set.iter())
        .copied()
        .filter(|s| {
            let in_base = base_set.contains(*s);
            let in_local = local_set.contains(*s);
            let in_remote = remote_set.contains(*s);
            // present in base → kept iff neither side removed it.
            // not in base → added by either side, keep.
            if in_base {
                in_local && in_remote
            } else {
                in_local || in_remote
            }
        })
        .collect();

    let mut index = Vec::new();
    let mut seen: HashSet<StableId> = HashSet::new();
    for s in remote.index.iter().chain(local.index.iter()) {
        if keep.contains(s) && seen.insert(s.clone()) {
            index.push(s.clone());
        }
    }

    let mut inbox: BTreeMap<String, StableId> = BTreeMap::new();
    let all_keys: BTreeSet<&String> = base
        .inbox
        .keys()
        .chain(local.inbox.keys())
        .chain(remote.inbox.keys())
        .collect();
    for k in all_keys {
        let in_base = base.inbox.contains_key(k);
        let lv = local.inbox.get(k);
        let rv = remote.inbox.get(k);
        if in_base {
            // Removal on either side wins; otherwise prefer remote on conflict.
            if let (Some(_), Some(rv)) = (lv, rv) {
                inbox.insert(k.clone(), rv.clone());
            }
        } else {
            // New on either side; remote wins on simultaneous-add conflict.
            if let Some(v) = rv.or(lv) {
                inbox.insert(k.clone(), v.clone());
            }
        }
    }

    let mut inbox_order = Vec::new();
    let mut seen_inbox = HashSet::new();
    for key in remote
        .inbox_order
        .iter()
        .chain(local.inbox_order.iter())
        .chain(base.inbox_order.iter())
    {
        if inbox.contains_key(key) && seen_inbox.insert(key.clone()) {
            inbox_order.push(key.clone());
        }
    }
    for key in inbox.keys() {
        if seen_inbox.insert(key.clone()) {
            inbox_order.push(key.clone());
        }
    }

    let can_pull = if local.can_pull == base.can_pull {
        remote.can_pull
    } else {
        local.can_pull
    };

    Queue {
        index,
        can_pull,
        inbox_order,
        inbox,
    }
}

/// After task refs are reconciled, copy every other fetched ref
/// (`refs/tsk-fetched/<remote>/{namespaces,queues,properties}/*`) onto its
/// `refs/tsk/*` counterpart with force-update. Better merging for these is
/// tracked separately (queue merge, namespace renumber, etc.).
pub fn fast_forward_non_task_refs(repo: &Repository, remote: &str) -> Result<()> {
    let prefix = fetched_prefix(remote);
    let names: Vec<String> = repo
        .references_glob(&format!("{prefix}*"))?
        .filter_map(|r| r.ok().and_then(|r| r.name().ok().map(String::from)))
        .collect();
    for name in names {
        let Some(rest) = name.strip_prefix(prefix.as_str()) else {
            continue;
        };
        if rest.starts_with("tasks/")
            || rest.starts_with("namespaces/")
            || rest.starts_with("queues/")
        {
            continue;
        }
        let Some(target) = repo.find_reference(&name).ok().and_then(|r| r.target()) else {
            continue;
        };
        let local_name = format!("refs/tsk/{rest}");
        repo.reference(&local_name, target, true, "pull")?;
    }
    Ok(())
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::object::{self, Task};
    use git2::{Signature, Time};
    use std::path::Path;

    fn init_repo(p: &Path) -> Repository {
        let r = Repository::init(p).unwrap();
        let mut cfg = r.config().unwrap();
        cfg.set_str("user.name", "Tester").unwrap();
        cfg.set_str("user.email", "t@e").unwrap();
        r
    }

    /// Set up a divergent pair of refs in one repo: local at refs/tsk/tasks/<s>
    /// and a "fetched-from-origin" tip at refs/tsk-fetched/origin/tasks/<s>.
    /// `local_props` and `remote_props` get applied to the same root content.
    fn make_diverged(
        repo: &Repository,
        content: &str,
        local_props: &[(&str, &str)],
        remote_props: &[(&str, &str)],
    ) -> StableId {
        let stable = object::create(repo, &Task::new(content), "create").unwrap();
        let root_oid = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        // Local edit.
        let mut t_local = Task::new(content);
        for (k, v) in local_props {
            t_local
                .properties
                .insert((*k).to_string(), vec![(*v).to_string()]);
        }
        object::update(repo, &stable, &t_local, "edit-local").unwrap();
        // Build remote commit branching off the root.
        let mut t_remote = Task::new(content);
        for (k, v) in remote_props {
            t_remote
                .properties
                .insert((*k).to_string(), vec![(*v).to_string()]);
        }
        let content_oid = repo.blob(t_remote.content.as_bytes()).unwrap();
        let mut tb = repo.treebuilder(None).unwrap();
        tb.insert("content", content_oid, 0o100644).unwrap();
        let title_oid = repo.blob(t_remote.title().as_bytes()).unwrap();
        tb.insert("title", title_oid, 0o100644).unwrap();
        for (k, vs) in &t_remote.properties {
            let body: String = vs.iter().map(|v| format!("{v}\n")).collect();
            let oid = repo.blob(body.as_bytes()).unwrap();
            tb.insert(k.as_str(), oid, 0o100644).unwrap();
        }
        let tree_oid = tb.write().unwrap();
        let sig = Signature::now("Remote", "r@x").unwrap();
        let parent = repo.find_commit(root_oid).unwrap();
        let remote_oid = repo
            .commit(
                None,
                &sig,
                &sig,
                "edit-remote",
                &repo.find_tree(tree_oid).unwrap(),
                &[&parent],
            )
            .unwrap();
        repo.reference(
            &format!("refs/tsk-fetched/origin/tasks/{}", stable.0),
            remote_oid,
            true,
            "test-setup",
        )
        .unwrap();
        stable
    }

    fn commit_task_at(
        repo: &Repository,
        parent: Oid,
        task: &Task,
        message: &str,
        author: &str,
        seconds: i64,
    ) -> Oid {
        let content_oid = repo.blob(task.content.as_bytes()).unwrap();
        let mut tb = repo.treebuilder(None).unwrap();
        tb.insert("content", content_oid, 0o100644).unwrap();
        for (k, vs) in &task.properties {
            let body = crate::propvalue::encode(vs);
            let oid = repo.blob(&body).unwrap();
            tb.insert(k.as_str(), oid, 0o100644).unwrap();
        }
        let tree_oid = tb.write().unwrap();
        let sig = Signature::new(author, "test@example.com", &Time::new(seconds, 0)).unwrap();
        let parent = repo.find_commit(parent).unwrap();
        repo.commit(
            None,
            &sig,
            &sig,
            message,
            &repo.find_tree(tree_oid).unwrap(),
            &[&parent],
        )
        .unwrap()
    }

    #[test]
    fn merge_clean_when_edits_dont_overlap() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        let stable = make_diverged(
            &repo,
            "shared",
            &[("priority", "high")],
            &[("status", "urgent")],
        );
        let recs = reconcile_task_refs(&repo, "origin", Strategy::Merge).unwrap();
        assert_eq!(recs.len(), 1);
        assert_eq!(recs[0].kind, ReconKind::Merged);
        // Merge commit has two parents.
        let head = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        let merge = repo.find_commit(head).unwrap();
        assert_eq!(merge.parent_count(), 2);
        // Both property changes survived.
        let task = object::read(&repo, &stable).unwrap().unwrap();
        assert_eq!(task.properties.get("priority").unwrap(), &vec!["high"]);
        assert_eq!(task.properties.get("status").unwrap(), &vec!["urgent"]);
    }

    #[test]
    fn rebase_replays_local_on_remote_preserving_authors() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        let stable = make_diverged(
            &repo,
            "shared",
            &[("priority", "high")],
            &[("status", "urgent")],
        );
        let recs = reconcile_task_refs(&repo, "origin", Strategy::Rebase).unwrap();
        assert_eq!(recs[0].kind, ReconKind::Rebased);
        // Rebased tip should be a single-parent commit whose parent chain
        // traces back through the remote's edit.
        let head = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        let tip = repo.find_commit(head).unwrap();
        assert_eq!(tip.parent_count(), 1);
        // Author of the rebased tip preserved (Tester from the local edit).
        assert_eq!(tip.author().name().unwrap(), "Tester");
        // Parent is the remote commit, authored by "Remote".
        let parent = tip.parent(0).unwrap();
        assert_eq!(parent.author().name().unwrap(), "Remote");
        let task = object::read(&repo, &stable).unwrap().unwrap();
        assert_eq!(task.properties.get("priority").unwrap(), &vec!["high"]);
        assert_eq!(task.properties.get("status").unwrap(), &vec!["urgent"]);
    }

    fn make_content_conflict(repo: &Repository) -> (StableId, Oid, Oid) {
        let stable = object::create(repo, &Task::new("v0"), "create").unwrap();
        let root_oid = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        let t_local = Task::new("v-local");
        object::update(repo, &stable, &t_local, "edit-local").unwrap();
        let local_oid = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        let t_remote = Task::new("v-remote");
        let remote_oid = commit_task_at(repo, root_oid, &t_remote, "edit-remote", "Remote", 200);
        repo.reference(
            &format!("refs/tsk-fetched/origin/tasks/{}", stable.0),
            remote_oid,
            true,
            "test",
        )
        .unwrap();
        (stable, local_oid, remote_oid)
    }

    #[test]
    fn content_conflict_commits_editor_result() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        let (stable, local_oid, remote_oid) = make_content_conflict(&repo);
        let kind = merge_strategy_inner(
            &repo,
            &stable,
            local_oid,
            remote_oid,
            namespace::DEFAULT_NS,
            |initial| {
                assert!(initial.contains("<<<<<<< ours\nv-local\n"));
                assert!(initial.contains("||||||| base\nv0\n"));
                assert!(initial.contains("=======\nv-remote\n"));
                assert!(initial.contains(">>>>>>> theirs\n"));
                Ok(Some("resolved title\n\nresolved body\n".into()))
            },
        )
        .unwrap();
        assert_eq!(kind, ReconKind::Conflict);
        let head = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        let merge = repo.find_commit(head).unwrap();
        assert_eq!(merge.parent_count(), 2);
        let task = object::read(&repo, &stable).unwrap().unwrap();
        assert_eq!(task.content, "resolved title\n\nresolved body\n");
    }

    #[test]
    fn empty_content_conflict_resolution_leaves_local_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        let (stable, local_oid, remote_oid) = make_content_conflict(&repo);
        let kind = merge_strategy_inner(
            &repo,
            &stable,
            local_oid,
            remote_oid,
            namespace::DEFAULT_NS,
            |_initial| Ok(Some("   \n\t\n".into())),
        )
        .unwrap();
        assert_eq!(kind, ReconKind::Conflict);
        let head = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        assert_eq!(head, local_oid);
        let task = object::read(&repo, &stable).unwrap().unwrap();
        assert_eq!(task.content, "v-local");
    }

    #[test]
    fn property_conflict_chooses_newer_timestamp_and_reindexes() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        let stable = object::create(&repo, &Task::new("shared"), "create").unwrap();
        let root_oid = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();

        let mut local_task = Task::new("shared");
        local_task
            .properties
            .insert("priority".into(), vec!["local".into()]);
        let local_oid = commit_task_at(&repo, root_oid, &local_task, "edit-local", "Local", 100);
        repo.reference(&stable.refname(), local_oid, true, "test-local")
            .unwrap();

        let mut remote_task = Task::new("shared");
        remote_task
            .properties
            .insert("priority".into(), vec!["remote".into()]);
        let remote_oid =
            commit_task_at(&repo, root_oid, &remote_task, "edit-remote", "Remote", 200);
        repo.reference(
            &format!("refs/tsk-fetched/origin/tasks/{}", stable.0),
            remote_oid,
            true,
            "test-remote",
        )
        .unwrap();

        let recs = reconcile_task_refs(&repo, "origin", Strategy::Merge).unwrap();
        assert_eq!(recs[0].kind, ReconKind::Conflict);

        let task = object::read(&repo, &stable).unwrap().unwrap();
        let values = task.properties.get("priority").unwrap();
        assert_eq!(values, &vec!["remote"]);

        let indexed = properties::read(&repo, "priority").unwrap();
        assert_eq!(indexed.get(&stable), Some(values));

        let head = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        let merge = repo.find_commit(head).unwrap();
        let message = merge.message().unwrap();
        assert!(message.contains("property conflict priority: chose theirs by timestamp"));
        assert!(message.contains("ours=100, theirs=200"));
    }

    #[test]
    fn fast_forward_when_local_is_ancestor() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        let stable = object::create(&repo, &Task::new("v0"), "create").unwrap();
        let root_oid = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        // Build a remote with one extra commit on top of the root.
        let content_oid = repo.blob(b"v0").unwrap();
        let mut tb = repo.treebuilder(None).unwrap();
        tb.insert("content", content_oid, 0o100644).unwrap();
        tb.insert("title", repo.blob(b"v0").unwrap(), 0o100644)
            .unwrap();
        tb.insert("status", repo.blob(b"open\n").unwrap(), 0o100644)
            .unwrap();
        let tree_oid = tb.write().unwrap();
        let sig = Signature::now("Remote", "r@x").unwrap();
        let parent = repo.find_commit(root_oid).unwrap();
        let remote_oid = repo
            .commit(
                None,
                &sig,
                &sig,
                "edit-remote",
                &repo.find_tree(tree_oid).unwrap(),
                &[&parent],
            )
            .unwrap();
        repo.reference(
            &format!("refs/tsk-fetched/origin/tasks/{}", stable.0),
            remote_oid,
            true,
            "test",
        )
        .unwrap();
        let recs = reconcile_task_refs(&repo, "origin", Strategy::Merge).unwrap();
        assert_eq!(recs[0].kind, ReconKind::FastForward);
        let head = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        assert_eq!(head, remote_oid);
    }

    #[test]
    fn new_remote_task_is_imported() {
        let dir = tempfile::tempdir().unwrap();
        let repo = init_repo(dir.path());
        // No local task; stash one only at the fetched ref.
        let stable = object::create(&repo, &Task::new("foreign"), "create").unwrap();
        let oid = repo
            .find_reference(&stable.refname())
            .unwrap()
            .target()
            .unwrap();
        // Move the local ref away so only fetched exists.
        repo.find_reference(&stable.refname())
            .unwrap()
            .delete()
            .unwrap();
        repo.reference(
            &format!("refs/tsk-fetched/origin/tasks/{}", stable.0),
            oid,
            true,
            "test",
        )
        .unwrap();
        let recs = reconcile_task_refs(&repo, "origin", Strategy::Merge).unwrap();
        assert_eq!(recs[0].kind, ReconKind::NewRemote);
        assert!(repo.find_reference(&stable.refname()).is_ok());
    }
}