//! The data structures representing contents of a Pijul repository in
//! memory at specific point in time. This representation is used to
//! compute the order on the chunks of a file, possibly detecting
//! conflicts in the process.
use backend::*;
use Result;
use conflict;
use std::collections::{HashMap, HashSet};
use std::cmp::min;

use std;
use rand;

bitflags! {
    struct Flags: u8 {
        const LINE_HALF_DELETED = 4;
        const LINE_VISITED = 2;
        const LINE_ONSTACK = 1;
    }
}


/// The elementary datum in the representation of the repository state
/// at any given point in time. We need this structure (as opposed to
/// working directly on a branch) in order to add more data, such as
/// strongly connected component identifier, to each node.
#[derive(Debug)]
pub struct Line {
    /// The internal identifier of the line.
    pub key: Key<PatchId>,

    // The status of the line with respect to a dfs of
    // a graph it appears in. This is 0 or
    // `LINE_HALF_DELETED`.
    flags: Flags,
    children: usize,
    n_children: usize,
    index: usize,
    lowlink: usize,
    scc: usize,
}


impl Line {
    pub fn is_zombie(&self) -> bool {
        self.flags.contains(LINE_HALF_DELETED)
    }
}

/// A graph, representing the whole content of the repository state at
/// a point in time. The encoding is a "flat adjacency list", where
/// each vertex contains a index `children` and a number of children
/// `n_children`. The children of that vertex are then
/// `&g.children[children .. children + n_children]`.
#[derive(Debug)]
pub struct Graph {
    /// Array of all alive lines in the graph. Line 0 is a dummy line
    /// at the end, so that all nodes have a common successor
    pub lines: Vec<Line>,
    /// Edge + index of the line in the "lines" array above. "None"
    /// means "dummy line at the end", and corresponds to line number
    /// 0.
    children: Vec<(Option<Edge>, VertexId)>,
}

#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash, Ord, PartialOrd)]
struct VertexId(usize);

const DUMMY_VERTEX: VertexId = VertexId(0);

impl std::ops::Index<VertexId> for Graph {
    type Output = Line;
    fn index(&self, idx: VertexId) -> &Self::Output {
        self.lines.index(idx.0)
    }
}
impl std::ops::IndexMut<VertexId> for Graph {
    fn index_mut(&mut self, idx: VertexId) -> &mut Self::Output {
        self.lines.index_mut(idx.0)
    }
}

use std::io::Write;
use hex::ToHex;

impl Graph {
    fn children(&self, i: VertexId) -> &[(Option<Edge>, VertexId)] {
        let ref line = self[i];
        &self.children[line.children..line.children + line.n_children]
    }

    fn child(&self, i: VertexId, j: usize) -> &(Option<Edge>, VertexId) {
        &self.children[self[i].children + j]
    }

    pub fn debug<W:Write>(&self, mut w: W) -> std::io::Result<()> {
        writeln!(w, "digraph {{")?;
        for (line, i) in self.lines.iter().zip(0..) {
            writeln!(w,
                     "n_{}[label=\"{}: {}\"];",
                     i, i, line.key.to_hex())?;
            for &(ref edge, VertexId(j)) in &self.children[line.children .. line.children + line.n_children] {
                if let Some(ref edge) = *edge {
                    writeln!(w,
                             "n_{}->n_{}[label=\"{:?} {}\"];",
                             i, j, edge.flag, edge.introduced_by.to_hex())?
                } else {
                    writeln!(w,
                             "n_{}->n_{}[label=\"none\"];",
                             i, j)?
                }
            }
        }
        writeln!(w, "}}")?;
        Ok(())
    }
}

use sanakirja::value::Value;
/// A "line outputter" trait.
pub trait LineBuffer<'a, T: 'a + Transaction> {
    fn output_line(&mut self, key: &Key<PatchId>, contents: Value<'a, T>) -> Result<()>;

    fn output_conflict_marker(&mut self, s: &'a str) -> Result<()>;
    fn begin_conflict(&mut self) -> Result<()> {
        self.output_conflict_marker(conflict::START_MARKER)
    }
    fn conflict_next(&mut self) -> Result<()> {
        self.output_conflict_marker(conflict::SEPARATOR)
    }
    fn end_conflict(&mut self) -> Result<()> {
        self.output_conflict_marker(conflict::END_MARKER)
    }
}

impl<'a, T: 'a + Transaction, W: std::io::Write> LineBuffer<'a, T> for W {
    fn output_line(&mut self, _: &Key<PatchId>, c: Value<T>) -> Result<()> {
        for chunk in c {
            try!(self.write_all(chunk))
        }
        Ok(())
    }

    fn output_conflict_marker(&mut self, s: &'a str) -> Result<()> {
        try!(self.write(s.as_bytes()));
        Ok(())
    }
}


#[derive(Clone, Debug)]
pub struct Visits {
    pub first: usize,
    pub last: usize,
    pub begins_conflict: bool,
    pub ends_conflict: bool,
    pub has_brothers: bool,
}

impl Default for Visits {
    fn default() -> Self {
        Visits { first: 0, last: 0, ends_conflict: false,
                 begins_conflict: false, has_brothers: false, }
    }
}

pub struct DFS {
    visits: Vec<Visits>,
    counter: usize,
    has_conflicts: bool,
}

impl DFS {
    pub fn new(n: usize) -> Self {
        DFS {
            visits: vec![Visits::default(); n],
            counter: 1,
            has_conflicts: false
        }
    }
}

impl DFS {
    fn mark_discovered(&mut self, scc: usize) {
        if self.visits[scc].first == 0 {
            self.visits[scc].first = self.counter;
            self.counter += 1;
        }
    }

    fn mark_last_visit(&mut self, scc: usize) {
        self.mark_discovered(scc);
        self.visits[scc].last = self.counter;
        self.counter += 1;
    }

    fn first_visit(&self, scc: usize) -> usize {
        self.visits[scc].first
    }

    fn last_visit(&self, scc: usize) -> usize {
        self.visits[scc].last
    }

    fn ends_conflict(&mut self, scc: usize) {
        self.visits[scc].ends_conflict = true
    }
    fn begins_conflict(&mut self, scc: usize) {
        self.visits[scc].begins_conflict = true;
        self.has_conflicts = true
    }
    fn has_brothers(&mut self, scc: usize) {
        self.visits[scc].has_brothers = true
    }
}

impl Graph {
    /*
    /// This is basically just Tarjan's strongly connected component algorithm.
    fn tarjan_dfs(&mut self,
                  scc: &mut Vec<Vec<VertexId>>,
                  stack: &mut Vec<VertexId>,
                  index: &mut usize,
                  n_l: VertexId) {
        {
            let ref mut l = self[n_l];
            debug!("tarjan: {:?}", l.key);
            (*l).index = *index;
            (*l).lowlink = *index;
            (*l).flags = (*l).flags | LINE_ONSTACK | LINE_VISITED;
            debug!("{:?} {:?} chi", (*l).key, (*l).n_children);
        }
        stack.push(n_l);
        *index = *index + 1;

        for i in 0..self[n_l].n_children {
            let &(_, n_child) = self.child(n_l, i);
            if !self[n_child].flags.contains(LINE_VISITED) {

                self.tarjan_dfs(scc, stack, index, n_child);
                self[n_l].lowlink = std::cmp::min(self[n_l].lowlink, self[n_child].lowlink);
            } else {
                if self[n_child].flags.contains(LINE_ONSTACK) {
                    self[n_l].lowlink = min(self[n_l].lowlink, self[n_child].index)
                }
            }
        }

        if self[n_l].index == self[n_l].lowlink {

            let mut v = Vec::new();
            loop {
                match stack.pop() {
                    None => break,
                    Some(n_p) => {
                        self[n_p].scc = scc.len();
                        self[n_p].flags = self[n_p].flags ^ LINE_ONSTACK;
                        v.push(n_p);
                        if n_p == n_l {
                            break;
                        }
                    }
                }
            }
            scc.push(v);
        }
    }
    */

    /// Tarjan's strongly connected component algorithm, returning a
    /// vector of strongly connected components, where each SCC is a
    /// vector of vertex indices.
    fn tarjan(&mut self) -> Vec<Vec<VertexId>> {
        if self.lines.len() <= 1 {
            return vec![vec![VertexId(0)]]
        }

        let mut call_stack = vec![(VertexId(1), 0, true)];

        let mut index = 0;
        let mut stack = Vec::new();
        let mut scc = Vec::new();
        while let Some((n_l, i, first_visit)) = call_stack.pop() {

            if first_visit {

                // First time we visit this node.
                let ref mut l = self[n_l];
                debug!("tarjan: {:?}", l.key);
                (*l).index = index;
                (*l).lowlink = index;
                (*l).flags = (*l).flags | LINE_ONSTACK | LINE_VISITED;
                debug!("{:?} {:?} chi", (*l).key, (*l).n_children);
                stack.push(n_l);
                index = index + 1;

            } else {

                let &(_, n_child) = self.child(n_l, i);
                self[n_l].lowlink = std::cmp::min(self[n_l].lowlink, self[n_child].lowlink);

            }

            let call_stack_length = call_stack.len();
            for j in i..self[n_l].n_children {
                let &(_, n_child) = self.child(n_l, j);
                if !self[n_child].flags.contains(LINE_VISITED) {

                    call_stack.push((n_l, j, false));
                    call_stack.push((n_child, 0, true));
                    break
                    // self.tarjan_dfs(scc, stack, index, n_child);
                } else {
                    if self[n_child].flags.contains(LINE_ONSTACK) {
                        self[n_l].lowlink = min(self[n_l].lowlink, self[n_child].index)
                    }
                }
            }
            if call_stack_length < call_stack.len() {
                // recursive call
                continue
            }
            // Here, all children of n_l have been visited.

            if self[n_l].index == self[n_l].lowlink {

                let mut v = Vec::new();
                loop {
                    match stack.pop() {
                        None => break,
                        Some(n_p) => {
                            self[n_p].scc = scc.len();
                            self[n_p].flags = self[n_p].flags ^ LINE_ONSTACK;
                            v.push(n_p);
                            if n_p == n_l {
                                break;
                            }
                        }
                    }
                }
                scc.push(v);
            }
        }
        scc
    }

    /// Run a depth-first search on this graph, assigning the
    /// `first_visit` and `last_visit` numbers to each node.
    fn dfs(&mut self,
               scc: &[Vec<VertexId>],
               dfs: &mut DFS,
               forward: &mut Vec<(Key<PatchId>, Edge)>) {

        let mut call_stack = vec![(scc.len()-1, HashSet::new(), true, None)];
        while let Some((n_scc, mut forward_scc, is_first_child, descendants)) = call_stack.pop() {

            debug!("dfs, n_scc = {:?}", n_scc);
            dfs.mark_discovered(n_scc);
            let mut descendants = if let Some(descendants) = descendants {
                descendants
            } else {

                // First visit / discovery of SCC n_scc.

                // After Tarjan's algorithm, the SCC numbers are in reverse
                // topological order.
                //
                // Here, we want to visit the first child in topological
                // order, hence the one with the largest SCC number first.
                //

                // Collect all descendants of this SCC, in order of increasing
                // SCC.
                let mut descendants = Vec::new();
                for cousin in scc[n_scc].iter() {

                    for &(_, n_child) in self.children(*cousin) {

                        let child_component = self[n_child].scc;
                        if child_component < n_scc {

                            // If this is a child and not a sibling.
                            descendants.push(child_component)

                        }
                    }
                }
                descendants.sort();
                debug!("sorted descendants: {:?}", descendants);
                descendants
            };

            // SCCs to which we have forward edges.
            let mut recursive_call = None;
            while let Some(child) = descendants.pop() {

                if dfs.first_visit(child) == 0 {

                    // This SCC has not yet been visited, visit it.
                    recursive_call = Some(child);
                    if !is_first_child {
                        // Two incomparable children, n_scc starts a conflict
                        debug!("!is_first_child, n_scc = {:?}", n_scc);
                        dfs.begins_conflict(n_scc);
                        dfs.has_brothers(child)
                    }
                    break

                } else if dfs.last_visit(child) != 0 && dfs.first_visit(child) > dfs.first_visit(n_scc) {
                    // This is a forward edge.
                    forward_scc.insert(child);

                } else if dfs.last_visit(child) < dfs.first_visit(n_scc) {
                    dfs.ends_conflict(child)
                }
            }
            if let Some(child) = recursive_call {
                call_stack.push((n_scc, forward_scc, false, Some(descendants)));
                call_stack.push((child, HashSet::new(), true, None));
            } else {
                dfs.mark_last_visit(n_scc);
                // After this, collect forward edges.
                for cousin in scc[n_scc].iter() {
                    for &(ref edge, n_child) in self.children(*cousin) {
                        if let Some(ref edge) = *edge {
                            if forward_scc.contains(&self[n_child].scc) && edge.flag.contains(PSEUDO_EDGE) {
                                forward.push((self[*cousin].key.clone(), edge.clone()))
                            }
                        }
                    }
                }
            }
        }
    }
}

impl<A: Transaction, R> T<A, R> {

    /// This function constructs a graph by reading the branch from the
    /// input key. It guarantees that all nodes but the first one (index
    /// 0) have a common descendant, which is index 0.
    pub fn retrieve<'a>(&'a self, branch: &Branch, key: &Key<PatchId>) -> Graph {

        let mut graph = Graph {
            lines: Vec::new(),
            children: Vec::new(),
        };
        // Insert last "dummy" line (so that all lines have a common descendant).
        graph.lines.push(Line {
            key: ROOT_KEY,
            flags: Flags::empty(),
            children: 0,
            n_children: 0,
            index: 0,
            lowlink: 0,
            scc: 0,
        });

        // Avoid the root key.
        let mut cache: HashMap<Key<PatchId>, VertexId> = HashMap::new();
        cache.insert(ROOT_KEY.clone(), DUMMY_VERTEX);
        let mut stack = vec![key.clone()];
        while let Some(key) = stack.pop() {
            let idx = VertexId(graph.lines.len());
            cache.insert(key.clone(), idx);

            debug!("{:?}", key);
            let is_zombie = {
                let mut tag = PARENT_EDGE | DELETED_EDGE;
                let mut is_zombie = false;

                // Find the first (k, v) after (key, tag).
                let first_edge = Edge::zero(tag);
                if let Some((k, v)) = self.iter_nodes(&branch, Some((&key, Some(&first_edge))))
                    .next() {
                        if *k == key && v.flag == tag {
                            is_zombie = true
                        }
                    }
                if !is_zombie {
                    tag = PARENT_EDGE | DELETED_EDGE | FOLDER_EDGE;
                    let first_edge = Edge::zero(tag);
                    if let Some((k, v)) =
                        self.iter_nodes(&branch, Some((&key, Some(&first_edge)))).next() {
                            if *k == key && v.flag == tag {
                                is_zombie = true
                            }
                        }
                }
                is_zombie
            };
            debug!("is_zombie: {:?}", is_zombie);
            let mut l = Line {
                key: key.clone(),
                flags: if is_zombie {
                    LINE_HALF_DELETED
                } else {
                    Flags::empty()
                },
                children: graph.children.len(),
                n_children: 0,
                index: 0,
                lowlink: 0,
                scc: 0,
            };

            for (_, v) in self.iter_nodes(&branch, Some((&key, None)))
                .take_while(|&(k, v)| *k == key && v.flag <= PSEUDO_EDGE | FOLDER_EDGE) {

                    debug!("-> v = {:?}", v);
                    graph.children.push((Some(v.clone()), DUMMY_VERTEX));
                    l.n_children += 1;
                    if ! cache.contains_key(&v.dest) {
                        stack.push(v.dest.clone())
                    } else {
                        debug!("v already visited");
                    }
                }
            // If this key has no children, give it the dummy child.
            if l.n_children == 0 {
                graph.children.push((None, DUMMY_VERTEX));
                l.n_children = 1;
            }
            graph.lines.push(l)
        }
        for &mut (ref child_key, ref mut child_idx) in graph.children.iter_mut() {
            if let Some(ref child_key) = *child_key {
                if let Some(idx) = cache.get(&child_key.dest) {
                    *child_idx = *idx
                }
            }
        }
        graph
    }

}

/// The conflict markers keep track of the number of conflicts, and is
/// used for outputting conflicts to a given LineBuffer.
///
/// "Zombie" conflicts are those conflicts introduced by zombie
/// vertices in the contained Graph.
struct ConflictMarkers<'b> {
    current_is_zombie: bool,
    current_conflicts: usize,
    graph: &'b Graph,
}

impl<'b> ConflictMarkers<'b> {
    fn output_zombie_markers_if_needed<'a, A: Transaction+'a, B:LineBuffer<'a, A>>(
        &mut self, buf: &mut B, vertex: VertexId
    ) -> Result<()> {

        if self.graph[vertex].is_zombie() && !self.current_is_zombie {
            debug!("begin zombie conflict: vertex = {:?}", vertex);
            self.current_is_zombie = true;
            buf.begin_conflict()?;
        } else if self.current_is_zombie {
            // Zombie segment has ended
            self.current_is_zombie = false;
            buf.end_conflict()?;
        }
        Ok(())
    }

    fn begin_conflict<'a, A: Transaction+'a, B:LineBuffer<'a, A>>(&mut self, buf: &mut B) -> Result<()> {
        buf.begin_conflict()?;
        self.current_conflicts += 1;
        Ok(())
    }
    fn end_conflict<'a, A: Transaction+'a, B:LineBuffer<'a, A>>(&mut self, buf: &mut B) -> Result<()> {
        if self.current_conflicts > 0 {
            buf.end_conflict()?;
            self.current_conflicts -= 1;
        }
        Ok(())
    }
}

impl<'a, A: Transaction + 'a, R> T<A, R> {
    pub fn output_file<B: LineBuffer<'a, A>>(&'a self,
                                             buf: &mut B,
                                             graph: &mut Graph,
                                             forward: &mut Vec<(Key<PatchId>, Edge)>)
                                             -> Result<bool> {

        debug!("output_file");

        let scc = graph.tarjan(); // SCCs are given here in reverse order.
        info!("There are {} SCC", scc.len());

        let mut dfs = DFS::new(scc.len());
        graph.dfs(&scc, &mut dfs, forward);

        debug!("dfs done");

        let mut i = scc.len() - 1;

        let mut output_scc = HashSet::new();
        let mut conflicts = ConflictMarkers {
            current_is_zombie: false,
            current_conflicts: 0,
            graph: &graph
        };

        loop {

            if dfs.visits[i].ends_conflict {
                conflicts.end_conflict(buf)?;
            } else if dfs.visits[i].has_brothers {
                buf.conflict_next()?;
            }

            if scc[i].len() > 1 {
                debug!("cycle conflict: {:?}", scc);
                let mut is_first = true;
                for &vertex in scc[i].iter() {

                    // Output conflict markers for the SCC conflict
                    if is_first {
                        conflicts.begin_conflict(buf)?;
                        is_first = false;
                    } else {
                        buf.conflict_next()?;
                    }

                    // Conflict markers for the zombie line.
                    conflicts.output_zombie_markers_if_needed(buf, vertex)?;

                    // Output the line
                    let ref key = graph[vertex].key;
                    if *key != ROOT_KEY {
                        if let Some(cont) = self.get_contents(&key) {
                            try!(buf.output_line(&key, cont))
                        }
                    }
                }
                conflicts.end_conflict(buf)?;
            } else if scc[i].len() == 1 {
                conflicts.output_zombie_markers_if_needed(buf, scc[i][0])?;
                let ref key = graph[scc[i][0]].key;
                if *key != ROOT_KEY {
                    if let Some(cont) = self.get_contents(&key) {
                        try!(buf.output_line(&key, cont))
                    }
                }
            }
            output_scc.insert(i);

            if dfs.visits[i].begins_conflict {
                conflicts.begin_conflict(buf)?;
            }

            if i == 0 {
                break;
            } else {
                i -= 1
            }
        }
        debug!("/output_file");
        Ok(dfs.has_conflicts)
    }
}

impl<'env, A: rand::Rng> MutTxn<'env, A> {
    pub fn remove_redundant_edges(&mut self,
                                  branch: &mut Branch,
                                  forward: &Vec<(Key<PatchId>, Edge)>)
                                  -> Result<()> {

        for &(ref key, ref edge) in forward.iter() {

            try!(self.del_nodes(branch, key, Some(edge)));
            let mut reverse = Edge {
                dest: key.clone(),
                flag: edge.flag,
                introduced_by: edge.introduced_by.clone(),
            };
            reverse.flag.toggle(PARENT_EDGE);
            try!(self.del_nodes(branch, &edge.dest, Some(&reverse)));
        }
        Ok(())
    }
}