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

/// Checkpoint payload version 2 (spec §11): file graphs by reference.

pub const CHECKPOINT_INDEX_VERSION: u64 = 2;

/// A file graph encoded in fewer bytes stays inline in a version 2 index.

pub const INLINE_GRAPH_LIMIT: usize = 2048;

#[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,

}

/// The log's order of changes: newest first by clock, then by hash, so

/// it is total.

pub fn log_order(a: (&ChangeHash, &Hlc), b: (&ChangeHash, &Hlc)) -> std::cmp::Ordering {

    b.1.cmp(a.1).then(b.0.cmp(a.0))

}

impl ChangeMeta {

    /// In the flat layout of local indexes and caches ([`crate::flat`]).

    pub(crate) fn write_flat(&self, w: &mut crate::flat::Writer) {

        w.fixed(&self.change_id.0);

        w.count(self.deps.len());

        for d in &self.deps {

            w.object(d);

        }

        w.object(&self.author);

        w.u64(self.hlc.wall_ms);

        w.u32(self.hlc.logical);

        w.bytes(self.message.as_bytes());

        w.u8(self.kind.tag() as u8);

    }

    pub(crate) fn read_flat(r: &mut crate::flat::Reader) -> Result<ChangeMeta, ObjectError> {

        let change_id = crate::ids::ChangeId(r.fixed::<16>()?);

        let mut deps = Vec::new();

        for _ in 0..r.count()? {

            deps.push(r.object()?);

        }

        Ok(ChangeMeta {

            change_id,

            deps,

            author: r.object()?,

            hlc: Hlc {

                wall_ms: r.u64()?,

                logical: r.u32()?,

            },

            message: r.text()?,

            kind: ChangeKind::from_tag(r.u8()? as u64)?,

        })

    }

    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 {

    /// Shared between clones, like `files` and `deferred`: a state kept

    /// beside another that differs by a few changes shares the rest.

    changes: crate::fasthash::FastMap<ChangeHash, Arc<ChangeMeta>>,

    tree: TreeGraph,

    /// Shared between clones: a clone copies a graph only when an edit

    /// changes it (`Arc::make_mut`), so keeping several states that differ

    /// by a few changes costs little more than one.

    files: crate::fasthash::FastMap<NodeId, Arc<FileGraph>>,

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

    /// Memoised transitive dependencies, bounded by the hashes they hold

    /// (each set is the whole history below a change, so keeping one for

    /// every change asked about grew with the square of history: #25). A

    /// mutex, not a `RefCell`, so a loaded state can be shared across

    /// threads by a host that caches it.

    closure: std::sync::RwLock<ClosureMemo>,

    /// Each change's level: 0 with no dependencies, else one more than its

    /// highest dependency. A change can only depend on changes of lower

    /// level, which bounds the walk [`DepOracle::depends_on`] does.

    levels: std::sync::RwLock<crate::fasthash::FastMap<ChangeHash, u32>>,

    /// `None` for a full state. For a history-only state (loaded for its

    /// changes and tree), the file graphs not built yet, by node: `files`

    /// holds only those loaded with [`State::load_graphs`].

    deferred: Option<crate::fasthash::FastMap<NodeId, Arc<Deferred>>>,

}

use crate::objects::TextRef;

/// A file graph a history-only state has not built: its base from a

/// version 2 checkpoint, if any, and the edits of later changes in the

/// order they were applied (spec §11). Building it later gives the same

/// graph, since graph construction does not depend on order (§8).

///

/// Read from a state index, it stays in the index, unread, until first

/// used: a command that builds a few files reads and checks only their

/// histories (oxygen#94).

#[derive(Clone, Default)]

struct Deferred {

    /// The history: `raw` read and decoded, with `more` after it. An

    /// error when `raw` does not match its hash.

    body: std::sync::OnceLock<Result<DeferredBody, String>>,

    /// The history as an index holds it, not read yet.

    raw: Option<RawHistory>,

    /// Changes added since the index was read, kept beside `raw`, so that

    /// adding one neither reads nor decodes the history.

    more: Vec<(ChangeHash, Vec<Edit>)>,

    more_texts: Vec<TextRef>,

    /// Its graph key, once worked out (or read from an index); reset by

    /// any edit added.

    key: std::sync::OnceLock<[u8; 32]>,

}

/// Where a state index's file histories are: the index file itself, read

/// on first use, or bytes in memory.

#[derive(Clone)]

pub enum Histories {

    Bytes(Arc<Vec<u8>>),

    /// The file, where its histories start, and how many bytes they take.

    File(Arc<std::fs::File>, u64, u64),

}

impl Histories {

    fn len(&self) -> u64 {

        match self {

            Histories::Bytes(b) => b.len() as u64,

            Histories::File(_, _, len) => *len,

        }

    }

    fn read(&self, at: u64, len: usize) -> std::io::Result<Vec<u8>> {

        match self {

            Histories::Bytes(b) => Ok(b[at as usize..at as usize + len].to_vec()),

            Histories::File(f, start, _) => {

                let mut buf = vec![0u8; len];

                read_exact_at(f, &mut buf, start + at)?;

                Ok(buf)

            }

        }

    }

}

#[cfg(unix)]

fn read_exact_at(f: &std::fs::File, buf: &mut [u8], at: u64) -> std::io::Result<()> {

    std::os::unix::fs::FileExt::read_exact_at(f, buf, at)

}

#[cfg(windows)]

fn read_exact_at(f: &std::fs::File, mut buf: &mut [u8], mut at: u64) -> std::io::Result<()> {

    while !buf.is_empty() {

        match std::os::windows::fs::FileExt::seek_read(f, buf, at)? {

            0 => return Err(std::io::ErrorKind::UnexpectedEof.into()),

            n => {

                buf = &mut buf[n..];

                at += n as u64;

            }

        }

    }

    Ok(())

}

/// One file's history in a state index: where it is and its hash.

#[derive(Clone)]

struct RawHistory {

    from: Histories,

    at: u64,

    len: usize,

    sum: [u8; 32],

}

impl RawHistory {

    /// The history's bytes, checked against the hash the index holds.

    fn bytes(&self) -> Result<Vec<u8>, ObjectError> {

        let damaged = |why: String| ObjectError::DamagedIndex(why);

        let b = self

            .from

            .read(self.at, self.len)

            .map_err(|e| damaged(format!("a file's history cannot be read: {e}")))?;

        if *blake3::hash(&b).as_bytes() != self.sum {

            return Err(damaged("a file's history does not match its hash".into()));

        }

        Ok(b)

    }

}

/// The closures kept, and how many hashes they hold in all.

#[derive(Default)]

struct ClosureMemo {

    sets: HashMap<ChangeHash, Arc<HashSet<ChangeHash>>>,

    held: usize,

}

/// The most hashes the closure memo holds before it starts again: a few

/// hundred megabytes at most.

const CLOSURE_MEMO_HASHES: usize = 4_000_000;

#[derive(Clone, Default)]

struct DeferredBody {

    base: Option<DeferredBase>,

    edits: Vec<(ChangeHash, Vec<Edit>)>,

    /// The text blob and line count of each insert in `edits` left without

    /// its lines, in order: read when the graph is built.

    texts: Vec<TextRef>,

}

impl Deferred {

    fn from_body(body: DeferredBody) -> Deferred {

        Deferred {

            body: std::sync::OnceLock::from(Ok(body)),

            ..Deferred::default()

        }

    }

    /// The history, read and checked on first use.

    fn body(&self) -> Result<&DeferredBody, ObjectError> {

        self.body

            .get_or_init(|| {

                let mut b = match &self.raw {

                    // Matching its hash, it is what `write_index` wrote: it

                    // decodes.

                    Some(raw) => raw

                        .bytes()

                        .and_then(|b| State::read_deferred_body(&b))

                        .map_err(|e| e.to_string())?,

                    None => DeferredBody::default(),

                };

                b.edits.extend(self.more.iter().cloned());

                b.texts.extend(self.more_texts.iter().cloned());

                Ok(b)

            })

            .as_ref()

            .map_err(|e| ObjectError::DamagedIndex(e.clone()))

    }

    /// Add a change's edits to the history: beside the index's encoding

    /// while it is unread, so that nothing is read or decoded to add them.

    fn add(&mut self, h: ChangeHash, edits: Vec<Edit>, texts: Vec<TextRef>) {

        if self.raw.is_some() && !matches!(self.body.get(), Some(Ok(_))) {

            self.more.push((h, edits));

            self.more_texts.extend(texts);

            // A body found damaged stays so: what was added cannot mend it.

            return;

        }

        // No encoding to read, or decoded already: this cannot fail.

        let b = self.body_mut().expect("a history in memory");

        b.edits.push((h, edits));

        b.texts.extend(texts);

    }

    fn body_mut(&mut self) -> Result<&mut DeferredBody, ObjectError> {

        self.body()?;

        self.raw = None;

        self.more.clear();

        self.more_texts.clear();

        Ok(self

            .body

            .get_mut()

            .and_then(|b| b.as_mut().ok())

            .expect("decoded above"))

    }

    fn into_body(mut self) -> Result<DeferredBody, ObjectError> {

        self.body()?;

        Ok(self

            .body

            .take()

            .and_then(|b| b.ok())

            .expect("decoded above"))

    }

}

#[derive(Clone)]

enum DeferredBase {

    Inline(Value),

    Blob(ObjectRef),

}

/// A local index of a history-only state (oxygen#41), in the flat layout

/// of [`crate::flat`]: its change metadata, tree (with its view), replace

/// records and every file's deferred history with its graph key, so a

/// command reads it rather than replaying history. Derived only: written

/// from a state loaded the long way, and read back as the same state.

impl State {

    const INDEX_MAGIC: &'static [u8] = b"oxy state index 4\n";

    /// The index of this state, or `None` when it has anything an index

    /// does not hold: a built graph, or text in memory (a change added

    /// whole rather than as its skeleton).

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

        use crate::flat::Writer;

        let deferred = self.deferred.as_ref()?;

        if !self.files.is_empty() {

            return None;

        }

        // The changes first, as a block a reader can take on its own (see

        // [`State::read_index_changes`]).

        let mut w = Writer::default();

        w.fixed(Self::INDEX_MAGIC);

        // In log order, newest first (oxygen#94): the log takes them as

        // they come, with no map and no sort.

        let mut changes: Vec<(&ChangeHash, &Arc<ChangeMeta>)> = self.changes.iter().collect();

        changes.sort_unstable_by(|a, b| log_order((a.0, &a.1.hlc), (b.0, &b.1.hlc)));

        w.count(changes.len());

        for (h, m) in changes {

            w.object(h);

            m.write_flat(&mut w);

        }

        let changes_block = std::mem::take(&mut w);

        w.bytes(&changes_block.0);

        // The tree and the files' histories as blocks of their own, so a

        // reader decodes them side by side.

        let mut tw = Writer::default();

        self.tree.write_flat(&mut tw);

        w.bytes(&tw.0);

        let mut replaces: Vec<(&NodeId, &Vec<ReplaceRecord>)> = self.replaces.iter().collect();

        replaces.sort_by_key(|(n, _)| **n);

        w.count(replaces.len());

        for (n, recs) in replaces {

            w.fixed(&n.0);

            w.count(recs.len());

            // In any order: which record is live does not depend on it.

            let mut recs: Vec<&ReplaceRecord> = recs.iter().collect();

            recs.sort();

            for (c, old, new) in recs {

                w.object(c);

                w.opt_object(old);

                w.opt_object(new);

            }

        }

        let mut front = w;

        // The table: per file its key and where its history is, with the

        // history's hash, which is checked when it is read (oxygen#94).

        let mut table = Writer::default();

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

        let mut nodes: Vec<(&NodeId, &Arc<Deferred>)> = deferred.iter().collect();

        nodes.sort_by_key(|(n, _)| **n);

        table.count(nodes.len());

        for (n, d) in nodes {

            let key = match d.key.get() {

                Some(k) => *k,

                None => {

                    let mut h = blake3::Hasher::new();

                    h.update(b"oxy graph key 1\0");

                    h.update(&n.0);

                    let k = d.graph_key(h).ok()?;

                    let _ = d.key.set(k);

                    k

                }

            };

            // Unchanged since read: its bytes as they are, checked.

            let bytes = match &d.raw {

                Some(raw) if d.more.is_empty() => raw.bytes().ok()?,

                _ => Self::write_deferred_body(d.body().ok()?)?,

            };

            table.fixed(&n.0);

            table.fixed(&key);

            table.u64(histories.len() as u64);

            table.u32(u32::try_from(bytes.len()).ok()?);

            table.fixed(blake3::hash(&bytes).as_bytes());

            histories.extend_from_slice(&bytes);

        }

        front.bytes(&table.0);

        let mut out = Writer::default();

        out.u64(front.0.len() as u64);

        out.fixed(&front.0);

        out.fixed(&histories);

        Some(out.0)

    }

    /// One file's history as an index encodes it; `None` when it holds

    /// text in memory, which an index does not.

    fn write_deferred_body(d: &DeferredBody) -> Option<Vec<u8>> {

        use crate::flat::Writer;

        let atom = |w: &mut Writer, a: &AtomRef| match a {

            AtomRef::Start => w.u8(0),

            AtomRef::End => w.u8(1),

            AtomRef::Atom { change, index } => {

                w.u8(2);

                w.object(change);

                w.u32(*index);

            }

        };

        let mut w = Writer::default();

        match &d.base {

            None => w.u8(0),

            Some(DeferredBase::Blob(r)) => {

                w.u8(1);

                w.object(r);

            }

            Some(DeferredBase::Inline(v)) => {

                w.u8(2);

                w.bytes(&cbor::encode(v));

            }

        }

        // Each change's edits with the texts of its inserts, sorted by

        // change: a graph does not depend on the order changes applied

        // in, and the index is then the same however history was read.

        let mut texts = d.texts.iter();

        let mut per_change: Vec<(&ChangeHash, &Vec<Edit>, Vec<&TextRef>)> = Vec::new();

        for (c, edits) in &d.edits {

            let n = edits

                .iter()

                .filter(|e| matches!(e, Edit::Insert { lines, .. } if lines.is_empty()))

                .count();

            per_change.push((c, edits, texts.by_ref().take(n).collect()));

        }

        per_change.sort_by_key(|(c, _, _)| **c);

        w.count(per_change.len());

        for (c, edits, _) in &per_change {

            w.object(c);

            w.count(edits.len());

            for e in edits.iter() {

                match e {

                    Edit::Insert {

                        node,

                        after,

                        before,

                        lines,

                    } => {

                        if !lines.is_empty() {

                            return None;

                        }

                        w.u8(0);

                        w.fixed(&node.0);

                        atom(&mut w, after);

                        atom(&mut w, before);

                    }

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

                        w.u8(1);

                        w.fixed(&node.0);

                        w.count(atoms.len());

                        for r in atoms {

                            w.object(&r.change);

                            w.u32(r.from);

                            w.u32(r.to);