//! 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.

    /// Every atom on some path from `first` to `last`, both included, live

    /// or deleted (spec §8): the atoms a range of lines is made of, and

    /// those that were deleted from between them. Empty when `last` does

    /// not follow `first`.

    pub fn between(&self, first: &AtomKey, last: &AtomKey) -> Vec<AtomKey> {

        let (from, to) = (Vertex::Atom(*first), Vertex::Atom(*last));

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

        let mut stack = vec![from];

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

            if ahead.insert(v) {

                stack.extend(

                    self.succ

                        .get(&v)

                        .map(|e| e.as_slice())

                        .unwrap_or(&[])

                        .iter()

                        .copied(),

                );

            }

        }

        if !ahead.contains(&to) {

            return vec![];

        }

        let mut pred: HashMap<Vertex, Vec<Vertex>> = HashMap::default();

        for (u, es) in &self.succ {

            if ahead.contains(u) {

                for w in es.iter() {

                    pred.entry(*w).or_default().push(*u);

                }

            }

        }

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

        let mut stack = vec![to];

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

            if behind.insert(v) {

                stack.extend(pred.get(&v).into_iter().flatten().copied());

            }

        }

        let mut out: Vec<AtomKey> = ahead

            .intersection(&behind)

            .filter_map(|v| match v {

                Vertex::Atom(k) if self.atoms.contains_key(k) => Some(*k),

                _ => None,

            })

            .collect();

        out.sort();

        out

    }

    /// The changes that inserted or deleted any of `atoms`.

    pub fn touching(&self, atoms: &[AtomKey]) -> BTreeSet<ChangeHash> {

        let mut out = BTreeSet::new();

        for k in atoms {

            out.insert(k.0);

            if let Some(info) = self.atoms.get(k) {

                out.extend(info.deleted_by.iter().copied());

            }

        }

        out

    }

    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 direct = self.succ.get(&v).map(Edges::as_slice).unwrap_or_default();

        // The common case, a successor that is itself kept: nothing to

        // contract.

        if direct.iter().all(|s| kept.contains(s)) {

            return direct.iter().copied().collect();

        }

        let mut out = BTreeSet::new();

        let mut stack: Vec<Vertex> = direct.to_vec();

        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

    }

    /// The atoms in order when the graph is one chain from Start to End with

    /// nothing deleted: a file only ever inserted into in sequence, as every

    /// file of a fresh commit or a clone is. Such a graph has no conflict.

    /// An edge to End besides the chain's next atom (Start's own, or an

    /// append's anchor) orders nothing, since End is last anyway.

    fn as_chain(&self) -> Option<Vec<AtomKey>> {

        if self

            .atoms

            .values()

            .any(|a| a.content.is_none() || !a.deleted_by.is_empty())

        {

            return None;

        }

        let mut order = Vec::with_capacity(self.atoms.len());

        let mut v = Vertex::Start;

        loop {

            let succ = self.succ.get(&v)?.as_slice();

            let mut atoms = succ.iter().filter(|s| **s != Vertex::End);

            match (atoms.next(), atoms.next()) {

                (None, _) if succ.contains(&Vertex::End) => break,

                (Some(Vertex::Atom(k)), None) if order.len() < self.atoms.len() => {

                    order.push(*k);

                    v = Vertex::Atom(*k);

                }

                _ => return None,

            }

        }

        (order.len() == self.atoms.len()).then_some(order)

    }

    /// Order the live atoms and locate conflicts.

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

        if let Some(order) = self.as_chain() {

            return Materialized {

                segments: if order.is_empty() {

                    vec![]

                } else {

                    vec![Segment::Lines(order)]

                },

            };

        }

        self.materialize_general(deps)

    }

    /// [`FileGraph::materialize`] for any graph: contraction, then a

    /// topological order with conflict regions.

    fn materialize_general(&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() {