//! A state: a dependency-closed set of changes, applied to a tree graph and

//! per-node file graphs (spec §8.3), with materialization (spec §9.5, §11).

use crate::cbor::{self, Value};

use crate::graph::{AtomKey, DepOracle, FileGraph, Materialized, Vertex};

use crate::hash::KeyId;

use crate::hash::{ChangeHash, ObjectRef};

use crate::ids::NodeId;

use crate::objects::{

    self as o, AtomRef, Change, ChangeKind, Edit, Hlc, Mode, NodeKind, ObjectError, TreeOp,

};

use crate::tree::{TreeGraph, TreeView};

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

use std::sync::{Arc, Mutex};

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]

pub enum StateError {

    #[error("change {1} depends on {0}, which is not in the state")]

    MissingDep(ChangeHash, ChangeHash),

}

/// One `Replace` edit: `(change, old blob, new blob)`.

type ReplaceRecord = (ChangeHash, Option<ObjectRef>, Option<ObjectRef>);

/// What a state remembers about a change: everything but its operations,

/// which live in the graphs once applied (and in the store).

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

pub struct ChangeMeta {

    pub change_id: crate::ids::ChangeId,

    pub deps: Vec<ChangeHash>,

    pub author: KeyId,

    pub hlc: Hlc,

    pub message: String,

    pub kind: ChangeKind,

}

impl ChangeMeta {

    pub fn of(c: &Change) -> ChangeMeta {

        ChangeMeta {

            change_id: c.change_id,

            deps: c.deps.clone(),

            author: c.author,

            hlc: c.hlc,

            message: c.message.clone(),

            kind: c.kind,

        }

    }

}

#[derive(Default)]

pub struct State {

    changes: HashMap<ChangeHash, ChangeMeta>,

    tree: TreeGraph,

    files: HashMap<NodeId, FileGraph>,

    replaces: HashMap<NodeId, Vec<ReplaceRecord>>,

    /// Memoised transitive dependencies. A mutex, not a `RefCell`, so a

    /// loaded state can be shared across threads by a host that caches it.

    closure: Mutex<HashMap<ChangeHash, Arc<HashSet<ChangeHash>>>>,

}

impl Clone for State {

    fn clone(&self) -> State {

        State {

            changes: self.changes.clone(),

            tree: self.tree.clone(),

            files: self.files.clone(),

            replaces: self.replaces.clone(),

            closure: Mutex::new(self.closure.lock().unwrap().clone()),

        }

    }

}

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

pub enum FileContent {

    Text {

        bytes: Vec<u8>,

        materialized: Materialized,

    },

    /// Live blobs; more than one is a replace conflict (spec §9.3).

    Opaque {

        blobs: Vec<ObjectRef>,

    },

    Dir,

}

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

pub struct FileState {

    pub node: NodeId,

    pub kind: NodeKind,

    pub mode: Mode,

    pub content: FileContent,

}

impl FileState {

    pub fn is_conflicted(&self) -> bool {

        match &self.content {

            FileContent::Text { materialized, .. } => materialized.is_conflicted(),

            FileContent::Opaque { blobs } => blobs.len() > 1,

            FileContent::Dir => false,

        }

    }

}

/// See `State::conflict_regions`.

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

pub struct ConflictRegions {

    pub plain: Vec<(AtomKey, Vec<u8>)>,

    pub region_atoms: Vec<AtomKey>,

    pub contributors: Vec<ChangeHash>,

}

/// The whole tree of a state, materialized.

#[derive(Clone, Debug, Default)]

pub struct Snapshot {

    pub tree: TreeView,

    pub files: BTreeMap<String, FileState>,

}

impl Snapshot {

    pub fn is_conflicted(&self) -> bool {

        self.tree.is_conflicted() || self.files.values().any(|f| f.is_conflicted())

    }

}

fn vertex(r: AtomRef) -> Vertex {

    match r {

        AtomRef::Start => Vertex::Start,

        AtomRef::End => Vertex::End,

        AtomRef::Atom { change, index } => Vertex::Atom((change, index)),

    }

}

/// Replace self references in a tree op with the concrete change hash.

pub fn resolve_tree_op(op: &TreeOp, me: ChangeHash) -> TreeOp {

    match op {

        TreeOp::RemoveNode { node, placement } => TreeOp::RemoveNode {

            node: *node,

            placement: placement.resolve(me),

        },

        TreeOp::Move {

            node,

            old,

            parent,

            name,

        } => TreeOp::Move {

            node: *node,

            old: old.resolve(me),

            parent: *parent,

            name: name.clone(),

        },

        TreeOp::SetMode { node, old, mode } => TreeOp::SetMode {

            node: *node,

            old: old.resolve(me),

            mode: *mode,

        },

        add @ TreeOp::AddNode { .. } => add.clone(),

    }

}

impl State {

    pub fn new() -> Self {

        Self::default()

    }

    pub fn len(&self) -> usize {

        self.changes.len()

    }

    pub fn is_empty(&self) -> bool {

        self.changes.is_empty()

    }

    pub fn contains(&self, h: &ChangeHash) -> bool {

        self.changes.contains_key(h)

    }

    pub fn get(&self, h: &ChangeHash) -> Option<&ChangeMeta> {

        self.changes.get(h)

    }

    pub fn changes(&self) -> impl Iterator<Item = (&ChangeHash, &ChangeMeta)> {

        self.changes.iter()

    }

    pub fn tree(&self) -> &TreeGraph {

        &self.tree

    }

    pub fn file(&self, node: &NodeId) -> Option<&FileGraph> {

        self.files.get(node)

    }

    /// Add a change whose dependencies are already present.

    pub fn add(&mut self, change: Change) -> Result<ChangeHash, StateError> {

        let h = change.hash();

        if self.changes.contains_key(&h) {

            return Ok(h);

        }

        for d in &change.deps {

            if !self.changes.contains_key(d) {

                return Err(StateError::MissingDep(*d, h));

            }

        }

        Ok(self.add_unchecked(change))

    }

    /// Add a change in any order; call `validate` once all are in.

    pub fn add_unchecked(&mut self, change: Change) -> ChangeHash {

        let h = change.hash();

        if self.changes.contains_key(&h) {

            return h;

        }

        self.apply(h, &change);

        self.changes.insert(h, ChangeMeta::of(&change));

        self.closure.lock().unwrap().clear();

        h

    }

    // ------------------------------------------------------------ checkpoints (spec §11)

    /// Serialize the whole state: change metadata, tree, file graphs, replaces.

    pub fn checkpoint_payload(&self) -> Vec<u8> {

        let mut hashes: Vec<&ChangeHash> = self.changes.keys().collect();

        hashes.sort();

        let changes = Value::Array(

            hashes

                .into_iter()

                .map(|h| {

                    let m = &self.changes[h];

                    Value::Array(vec![

                        Value::Bytes(h.0.to_vec()),

                        Value::Bytes(m.change_id.0.to_vec()),

                        Value::Array(m.deps.iter().map(|d| Value::Bytes(d.0.to_vec())).collect()),

                        Value::Text(m.message.clone()),

                        Value::Uint(m.hlc.wall_ms),

                        Value::Uint(m.hlc.logical as u64),

                        Value::Uint(m.kind.tag()),

                        Value::Bytes(m.author.0.to_vec()),

                    ])

                })

                .collect(),

        );

        let mut nodes: Vec<&NodeId> = self.files.keys().collect();

        nodes.sort();

        let files = Value::Array(

            nodes

                .into_iter()

                .map(|n| Value::Array(vec![Value::Bytes(n.0.to_vec()), self.files[n].to_value()]))

                .collect(),

        );

        let mut rnodes: Vec<&NodeId> = self.replaces.keys().collect();

        rnodes.sort();

        let opt =

            |o: &Option<ObjectRef>| o.map(|r| Value::Bytes(r.0.to_vec())).unwrap_or(Value::Null);

        let replaces = Value::Array(

            rnodes

                .into_iter()

                .map(|n| {

                    Value::Array(vec![

                        Value::Bytes(n.0.to_vec()),

                        Value::Array(

                            self.replaces[n]

                                .iter()

                                .map(|(c, o, nw)| {

                                    Value::Array(vec![Value::Bytes(c.0.to_vec()), opt(o), opt(nw)])

                                })

                                .collect(),

                        ),

                    ])

                })

                .collect(),

        );

        cbor::encode(&Value::map([

            (0, Value::Uint(o::FORMAT_VERSION)),

            (1, changes),

            (2, self.tree.to_value()),

            (3, files),

            (4, replaces),

        ]))

    }

    pub fn from_checkpoint_payload(bytes: &[u8]) -> Result<State, ObjectError> {

        let v = cbor::decode(bytes)?;

        let m = o::map(&v, "checkpoint payload")?;

        o::check_version(m)?;

        let mut st = State::new();

        for c in o::array(o::field(m, 1, "changes")?, "changes")? {

            let e = o::array(c, "change meta")?;

            if e.len() != 8 {

                return Err(ObjectError::Schema("change meta: expected 8 fields".into()));

            }

            let hash = o::oref(&e[0], "hash")?;

            let meta = ChangeMeta {

                change_id: crate::ids::ChangeId::from_slice(o::bytes(&e[1], "change id")?)

                    .ok_or_else(|| ObjectError::Schema("change id: expected 16 bytes".into()))?,

                deps: o::array(&e[2], "deps")?

                    .iter()

                    .map(|d| o::oref(d, "dep"))

                    .collect::<Result<Vec<_>, _>>()?,

                message: o::text(&e[3], "message")?.to_owned(),

                hlc: Hlc {

                    wall_ms: o::uint(&e[4], "wall")?,

                    logical: o::u32_(&e[5], "logical")?,

                },

                kind: ChangeKind::from_tag(o::uint(&e[6], "kind")?)?,

                author: o::oref(&e[7], "author")?,

            };

            st.changes.insert(hash, meta);

        }

        st.tree = TreeGraph::from_value(o::field(m, 2, "tree")?)?;

        for f in o::array(o::field(m, 3, "files")?, "files")? {

            let e = o::array(f, "file")?;

            if e.len() != 2 {

                return Err(ObjectError::Schema("file: expected [node, graph]".into()));

            }

            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)

                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;

            st.files.insert(node, FileGraph::from_value(&e[1])?);

        }

        let opt = |v: &Value| -> Result<Option<ObjectRef>, ObjectError> {

            if v.is_null() {

                Ok(None)

            } else {

                o::oref(v, "blob").map(Some)

            }

        };

        for r in o::array(o::field(m, 4, "replaces")?, "replaces")? {

            let e = o::array(r, "replace")?;

            if e.len() != 2 {

                return Err(ObjectError::Schema(

                    "replace: expected [node, records]".into(),

                ));

            }

            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)

                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;

            let mut recs = Vec::new();

            for rec in o::array(&e[1], "records")? {

                let x = o::array(rec, "record")?;

                if x.len() != 3 {

                    return Err(ObjectError::Schema("record: expected 3 fields".into()));

                }

                recs.push((o::oref(&x[0], "change")?, opt(&x[1])?, opt(&x[2])?));

            }

            st.replaces.insert(node, recs);

        }

        Ok(st)

    }

    pub fn validate(&self) -> Result<(), StateError> {

        for (h, c) in &self.changes {

            for d in &c.deps {

                if !self.changes.contains_key(d) {

                    return Err(StateError::MissingDep(*d, *h));

                }

            }

        }

        Ok(())

    }

    fn apply(&mut self, h: ChangeHash, c: &Change) {

        for (i, op) in c.tree_ops.iter().enumerate() {

            self.tree.apply(h, i as u32, &resolve_tree_op(op, h));

        }

        let mut counters: HashMap<NodeId, u32> = HashMap::new();

        for e in &c.edits {

            match e {

                Edit::Insert {

                    node,

                    after,

                    before,

                    lines,

                } => {

                    let idx = counters.entry(*node).or_default();

                    let first = *idx;

                    *idx += lines.len() as u32;

                    self.files.entry(*node).or_default().insert(

                        h,

                        first,

                        vertex(after.resolve(h)),

                        vertex(before.resolve(h)),

                        lines,

                    );

                }

                Edit::Delete { node, atoms } => {

                    let g = self.files.entry(*node).or_default();

                    for r in atoms {

                        g.delete(h, &r.resolve(h));

                    }

                }

                Edit::Replace { node, old, new } => {

                    self.replaces

                        .entry(*node)

                        .or_default()

                        .push((h, *old, *new));

                }

            }

        }

    }

    /// Transitive dependencies of `h`, excluding `h` itself.

    pub fn closure(&self, h: &ChangeHash) -> Arc<HashSet<ChangeHash>> {

        if let Some(c) = self.closure.lock().unwrap().get(h) {

            return c.clone();

        }

        let mut out = HashSet::new();

        let mut stack: Vec<ChangeHash> = self

            .changes

            .get(h)

            .map(|c| c.deps.clone())

            .unwrap_or_default();

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

            if out.insert(d) {

                if let Some(c) = self.changes.get(&d) {

                    stack.extend(c.deps.iter().copied());

                }

            }

        }

        let rc = Arc::new(out);

        self.closure.lock().unwrap().insert(*h, rc.clone());

        rc

    }

    /// Changes that no other change depends on directly, sorted.

    pub fn heads(&self) -> Vec<ChangeHash> {

        let depended: HashSet<&ChangeHash> =

            self.changes.values().flat_map(|c| c.deps.iter()).collect();

        let mut heads: Vec<ChangeHash> = self

            .changes

            .keys()

            .filter(|h| !depended.contains(h))

            .copied()

            .collect();

        heads.sort();

        heads

    }

    /// Maximal elements among `hashes` (those not in the closure of another).

    pub fn maximal(&self, hashes: &[ChangeHash]) -> Vec<ChangeHash> {

        let set: BTreeSet<ChangeHash> = hashes.iter().copied().collect();

        set.iter()

            .filter(|h| !set.iter().any(|o| o != *h && self.closure(o).contains(h)))

            .copied()