//! The per-file atom graph (spec §7.3), its materialization, and text conflict

//! detection (spec §9.1, §9.5).

//!

//! The graph is a pure function of the set of changes applied to it: inserts

//! add vertices and edges, deletes set a flag, and nothing is ever removed.

use crate::fasthash::{FastMap as HashMap, FastSet as HashSet};

use crate::hash::ChangeHash;

use crate::objects::AtomRange;

use std::collections::{BTreeMap, BTreeSet};

/// `(introducing change, index within that change for this node)`.

pub type AtomKey = (ChangeHash, u32);

#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Debug)]

pub enum Vertex {

    Start,

    Atom(AtomKey),

    End,

}

/// Transitive dependency test between changes; `depends_on(a, a)` is true.

pub trait DepOracle {

    fn depends_on(&self, a: &ChangeHash, b: &ChangeHash) -> bool;

}

/// No change depends on any other: every pair commutes.

pub struct NoDeps;

impl DepOracle for NoDeps {

    fn depends_on(&self, a: &ChangeHash, b: &ChangeHash) -> bool {

        a == b

    }

}

#[derive(Default, Clone, Debug)]

struct AtomInfo {

    /// `None` until the introducing change has been applied.

    content: Option<Vec<u8>>,

    deleted_by: BTreeSet<ChangeHash>,

}

#[derive(Clone, Debug)]

pub struct FileGraph {

    atoms: HashMap<AtomKey, AtomInfo>,

    /// Every vertex, with its direct successors.

    succ: HashMap<Vertex, Edges>,

}

/// A vertex's direct successors, sorted and without repeats. Nearly every

/// vertex has one, which is kept inline without an allocation.

#[derive(Clone, Debug, Default, PartialEq, Eq)]

enum Edges {

    #[default]

    None,

    One(Vertex),

    Many(Vec<Vertex>),

}

impl Edges {

    fn insert(&mut self, v: Vertex) {

        match self {

            Edges::None => *self = Edges::One(v),

            Edges::One(a) if *a == v => {}

            Edges::One(a) => {

                let a = *a;

                *self = Edges::Many(if a < v { vec![a, v] } else { vec![v, a] });

            }

            Edges::Many(list) => {

                if let Err(i) = list.binary_search(&v) {

                    list.insert(i, v);

                }

            }

        }

    }

    fn as_slice(&self) -> &[Vertex] {

        match self {

            Edges::None => &[],

            Edges::One(v) => std::slice::from_ref(v),

            Edges::Many(list) => list,

        }

    }

    fn iter(&self) -> std::slice::Iter<'_, Vertex> {

        self.as_slice().iter()

    }

}

impl Default for FileGraph {

    fn default() -> Self {

        Self::new()

    }

}

/// One piece of a materialized file.

#[derive(Clone, Debug, PartialEq, Eq)]

pub enum Segment {

    Lines(Vec<AtomKey>),

    /// Order conflict: alternatives that cannot be ordered against each other.

    Conflict(Vec<Vec<AtomKey>>),

    /// Delete/insert conflict: `inserted` was anchored to `deleted` by a change

    /// that commutes with the deletion.

    Zombie {

        deleted: Vec<AtomKey>,

        inserted: Vec<AtomKey>,

    },

}

#[derive(Clone, Debug, PartialEq, Eq, Default)]

pub struct Materialized {

    pub segments: Vec<Segment>,

}

impl Materialized {

    pub fn is_conflicted(&self) -> bool {

        self.segments

            .iter()

            .any(|s| !matches!(s, Segment::Lines(_)))

    }

    pub fn conflict_count(&self) -> usize {

        self.segments

            .iter()

            .filter(|s| !matches!(s, Segment::Lines(_)))

            .count()

    }

    /// Every live atom in materialized order, conflict sides concatenated.

    pub fn live_atoms(&self) -> Vec<AtomKey> {

        let mut out = Vec::new();

        for s in &self.segments {

            match s {

                Segment::Lines(a) => out.extend(a),

                Segment::Conflict(sides) => sides.iter().for_each(|x| out.extend(x)),

                Segment::Zombie { inserted, .. } => out.extend(inserted),

            }

        }

        out

    }

}

impl FileGraph {

    pub fn new() -> Self {

        let mut g = FileGraph {

            atoms: HashMap::default(),

            succ: HashMap::default(),

        };

        g.edge(Vertex::Start, Vertex::End);

        g

    }

    fn edge(&mut self, a: Vertex, b: Vertex) {

        self.succ.entry(a).or_default().insert(b);

        self.succ.entry(b).or_default();

    }

    /// Apply an `Insert` by `change` whose first new atom has index `first_index`.

    pub fn insert(

        &mut self,

        change: ChangeHash,

        first_index: u32,

        after: Vertex,

        before: Vertex,

        lines: &[Vec<u8>],

    ) {

        self.insert_owned(change, first_index, after, before, lines.to_vec());

    }

    /// [`FileGraph::insert`] taking the lines, so loading a state moves

    /// them in rather than copying. Each new atom's entry is made once, as

    /// the source of the next edge, and the chain's ends are ensured last.

    pub fn insert_owned(

        &mut self,

        change: ChangeHash,

        first_index: u32,

        after: Vertex,

        before: Vertex,

        lines: Vec<Vec<u8>>,

    ) {

        self.atoms.reserve(lines.len());

        self.succ.reserve(lines.len() + 1);

        let mut prev = after;

        for (i, line) in lines.into_iter().enumerate() {

            let key = (change, first_index + i as u32);

            self.atoms.entry(key).or_default().content = Some(line);

            let v = Vertex::Atom(key);

            self.succ.entry(prev).or_default().insert(v);

            prev = v;

        }

        self.succ.entry(prev).or_default().insert(before);

        self.succ.entry(before).or_default();

    }

    /// Apply a `Delete` of `range` by change `by`.

    pub fn delete(&mut self, by: ChangeHash, range: &AtomRange) {

        for i in range.from..range.to {

            self.atoms

                .entry((range.change, i))

                .or_default()

                .deleted_by

                .insert(by);

        }

    }

    pub fn is_live(&self, k: &AtomKey) -> bool {

        self.atoms

            .get(k)

            .is_some_and(|a| a.content.is_some() && a.deleted_by.is_empty())

    }

    pub fn content(&self, k: &AtomKey) -> Option<&[u8]> {

        self.atoms.get(k).and_then(|a| a.content.as_deref())

    }

    pub fn deleters(&self, k: &AtomKey) -> impl Iterator<Item = &ChangeHash> {

        self.atoms

            .get(k)

            .into_iter()

            .flat_map(|a| a.deleted_by.iter())

    }

    /// Number of atoms introduced by `change` in this graph (for index bookkeeping).

    pub fn atom_count(&self) -> usize {

        self.atoms.len()

    }

    /// The changes that inserted a live atom directly after `k`. A change

    /// that deletes `k` must depend on them (spec §8.1), or `is_ghost`

    /// reads the deletion as concurrent with those insertions.

    pub fn live_successor_changes(&self, k: &AtomKey) -> Vec<ChangeHash> {

        self.succ

            .get(&Vertex::Atom(*k))

            .map(|ss| {

                ss.iter()

                    .filter_map(|s| match s {

                        Vertex::Atom(y) if self.is_live(y) => Some(y.0),

                        _ => None,

                    })

                    .collect()

            })

            .unwrap_or_default()

    }

    fn is_ghost(&self, k: &AtomKey, info: &AtomInfo, deps: &dyn DepOracle) -> bool {

        // A deleted atom is a ghost when a live direct successor was inserted

        // by a change that commutes with every deleter: neither depends on

        // the other (spec §9.1, delete/insert). If the inserter came after the

        // deletion, or the deleter saw the insertion, there is no conflict.

        self.succ.get(&Vertex::Atom(*k)).is_some_and(|ss| {

            ss.iter().any(|s| match s {

                Vertex::Atom(y) if self.is_live(y) => !info

                    .deleted_by

                    .iter()

                    .any(|d| deps.depends_on(d, &y.0) || deps.depends_on(&y.0, d)),

                _ => false,

            })

        })

    }

    /// Successors of `v` in the graph with non-kept vertices contracted away.

    fn reduced_succ(

        &self,

        v: Vertex,

        kept: &HashSet<Vertex>,

        memo: &mut HashMap<Vertex, BTreeSet<Vertex>>,

    ) -> BTreeSet<Vertex> {

        let mut out = BTreeSet::new();

        let mut stack: Vec<Vertex> = self

            .succ

            .get(&v)

            .map(|s| s.iter().copied().collect())

            .unwrap_or_default();

        let mut seen = HashSet::default();

        while let Some(s) = stack.pop() {

            if kept.contains(&s) {

                out.insert(s);

            } else if seen.insert(s) {

                if let Some(m) = memo.get(&s) {

                    out.extend(m.iter().copied());

                } else if let Some(next) = self.succ.get(&s) {

                    stack.extend(next.iter().copied());

                }

            }

        }

        out

    }

    /// Order the live atoms and locate conflicts.

    pub fn materialize(&self, deps: &dyn DepOracle) -> Materialized {

        // 1. Vertices that survive contraction: live atoms, ghosts, Start, End.

        let mut kept: HashSet<Vertex> = HashSet::default();

        let mut ghosts: HashSet<AtomKey> = HashSet::default();

        kept.insert(Vertex::Start);

        kept.insert(Vertex::End);

        for (k, info) in &self.atoms {

            if info.content.is_none() {

                continue;

            }

            if info.deleted_by.is_empty() {

                kept.insert(Vertex::Atom(*k));

            } else if self.is_ghost(k, info, deps) {

                kept.insert(Vertex::Atom(*k));

                ghosts.insert(*k);

            }

        }

        // 2. Reduced graph over kept vertices.

        let mut memo = HashMap::default();

        let mut rsucc: HashMap<Vertex, BTreeSet<Vertex>> = HashMap::default();

        let mut indeg: HashMap<Vertex, usize> = HashMap::default();

        for &u in &kept {

            indeg.entry(u).or_insert(0);

            let rs = self.reduced_succ(u, &kept, &mut memo);

            for &w in &rs {

                *indeg.entry(w).or_insert(0) += 1;

            }

            rsucc.insert(u, rs);

        }

        // 3. Kahn's algorithm with side labelling for conflict regions.

        let mut ready: BTreeSet<Vertex> = indeg

            .iter()

            .filter(|(_, &d)| d == 0)

            .map(|(v, _)| *v)

            .collect();

        let mut side: HashMap<Vertex, u32> = HashMap::default();

        let mut joins: HashSet<Vertex> = HashSet::default();

        let mut next_side = 0u32;

        let mut plain: Vec<Vertex> = Vec::new();

        let mut conflict: Option<BTreeMap<u32, Vec<Vertex>>> = None;

        let mut segments: Vec<Segment> = Vec::new();

        let flush_plain = |plain: &mut Vec<Vertex>, segments: &mut Vec<Segment>| {

            let atoms: Vec<AtomKey> = plain

                .iter()

                .filter_map(|v| {

                    if let Vertex::Atom(k) = v {

                        Some(*k)

                    } else {

                        None

                    }

                })

                .collect();

            plain.clear();

            if !atoms.is_empty() {

                segments.extend(split_zombies(atoms, &ghosts));

            }

        };

        let finish_conflict = |c: BTreeMap<u32, Vec<Vertex>>, segments: &mut Vec<Segment>| {

            let sides: Vec<Vec<AtomKey>> = c

                .into_values()

                .map(|vs| {

                    vs.into_iter()

                        .filter_map(|v| match v {

                            Vertex::Atom(k) if !ghosts.contains(&k) => Some(k),

                            _ => None,

                        })

                        .collect::<Vec<_>>()

                })

                .filter(|s| !s.is_empty())

                .collect();

            match sides.len() {

                0 => {}

                1 => segments.push(Segment::Lines(sides.into_iter().next().unwrap())),

                _ => segments.push(Segment::Conflict(sides)),

            }

        };

        while !ready.is_empty() {

            if ready.len() >= 2 {

                if conflict.is_none() {

                    flush_plain(&mut plain, &mut segments);

                    conflict = Some(BTreeMap::new());

                }

                for r in &ready {

                    if !side.contains_key(r) {

                        side.insert(*r, next_side);

                        next_side += 1;

                    }

                }

            }

            let v = ready.pop_first().unwrap();

            let my_side = side.get(&v).copied();

            match (&mut conflict, my_side) {

                (Some(c), Some(s)) => c.entry(s).or_default().push(v),

                (Some(_), None) => {

                    let c = conflict.take().unwrap();

                    finish_conflict(c, &mut segments);

                    plain.push(v);

                }

                (None, _) => plain.push(v),

            }

            for &w in &rsucc[&v] {

                let d = indeg.get_mut(&w).unwrap();

                *d -= 1;

                if conflict.is_some() && !joins.contains(&w) {

                    match (my_side, side.get(&w).copied()) {

                        (Some(s), None) => {

                            side.insert(w, s);

                        }

                        (Some(s), Some(t)) if s != t => {

                            side.remove(&w);

                            joins.insert(w);

                        }

                        (None, Some(_)) => {

                            side.remove(&w);

                            joins.insert(w);

                        }

                        _ => {}

                    }

                }

                if *d == 0 {

                    ready.insert(w);

                }

            }

        }

        if let Some(c) = conflict.take() {

            finish_conflict(c, &mut segments);

        }

        flush_plain(&mut plain, &mut segments);

        Materialized { segments }

    }

    /// Render a materialization to bytes with conflict markers (spec §9.5).

    /// `label` gives the short description of a change for marker headers.

    pub fn render(&self, m: &Materialized, label: &dyn Fn(&ChangeHash) -> String) -> Vec<u8> {

        let total = m.conflict_count();

        let mut n = 0;

        let mut out: Vec<u8> = Vec::new();

        let push_atoms = |out: &mut Vec<u8>, atoms: &[AtomKey]| {

            for a in atoms {

                if let Some(c) = self.content(a) {

                    out.extend_from_slice(c);

                }

            }

        };

        let newline_if_needed = |out: &mut Vec<u8>| {

            if !out.is_empty() && *out.last().unwrap() != b'\n' {

                out.push(b'\n');

            }

        };

        for seg in &m.segments {

            match seg {

                Segment::Lines(a) => push_atoms(&mut out, a),

                Segment::Conflict(sides) => {

                    n += 1;

                    let k = sides.len();

                    for (i, s) in sides.iter().enumerate() {

                        newline_if_needed(&mut out);

                        let who = s.first().map(|a| label(&a.0)).unwrap_or_default();

                        let marker = if i == 0 { "<<<<<<<" } else { "=======" };

                        out.extend_from_slice(

                            format!("{marker} oxy conflict {n} of {total}, side {} of {k}: change {who}\n", i + 1).as_bytes(),