//! Checkpoints (spec §11): a serialized state kept per bookmark so loading

//! a state costs the changes since the checkpoint rather than all of them.

//!

//! The newest checkpoint of each bookmark is pointed at by

//! `.oxy/checkpoints/<bookmark>`; older ones become unreachable and are

//! garbage collected. A checkpoint applies to a requested head set when all

//! of its heads lie in that set's closure, which the loader establishes while

//! walking from the requested heads and stopping at checkpoint members.

use super::*;

use crate::objects::Checkpoint;

impl Repo {

    fn checkpoint_pointer(&self, bookmark: &str) -> PathBuf {

        self.oxy.join("checkpoints").join(name_file(bookmark))

    }

    /// The newest checkpoint recorded for a bookmark.

    pub fn latest_checkpoint(&self, bookmark: &str) -> Result<Option<(ObjectRef, Checkpoint)>> {

        match fs::read_to_string(self.checkpoint_pointer(bookmark)) {

            Ok(h) => {

                let r = ObjectRef::from_hex(h.trim())

                    .ok_or_else(|| RepoError::Msg("corrupt checkpoint pointer".into()))?;

                match self.store.get_typed(&r, ObjectType::Checkpoint) {

                    Ok(bytes) => Ok(Some((r, Checkpoint::decode(&bytes)?))),

                    Err(StoreError::NotFound(_)) => Ok(None),

                    Err(e) => Err(e.into()),

                }

            }

            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),

            Err(e) => Err(e.into()),

        }

    }

    /// Every checkpoint pointer, for garbage collection roots.

    pub(crate) fn checkpoint_refs(&self) -> Result<Vec<ObjectRef>> {

        let dir = self.oxy.join("checkpoints");

        let mut out = Vec::new();

        if let Ok(rd) = fs::read_dir(&dir) {

            for e in rd.flatten() {

                if let Ok(h) = fs::read_to_string(e.path()) {

                    if let Some(r) = ObjectRef::from_hex(h.trim()) {

                        out.push(r);

                    }

                }

            }

        }

        Ok(out)

    }

    /// Write a checkpoint of `state` (whose heads are `heads`) for `bookmark`.

    pub(crate) fn write_checkpoint(

        &self,

        bookmark: &str,

        state: &State,

        heads: &[ChangeHash],

    ) -> Result<(ObjectRef, usize)> {

        let batch = self.store.begin_batch()?;

        let (blob, payload_len) = self.store_checkpoint_payload(state)?;

        let cp = Checkpoint {

            heads: StateObject::new(heads.to_vec()).heads,

            count: state.len() as u64,

            payload: blob,

        };

        let r = self.store.put(ObjectType::Checkpoint, &cp.encode())?;

        batch.finish()?;

        fs::create_dir_all(self.oxy.join("checkpoints"))?;

        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())?;

        Ok((r, payload_len))

    }

    /// Store `state` as a version 2 payload (spec §11): the graphs of large

    /// files as blobs of their own, shared with earlier checkpoints when

    /// unchanged, and the index naming them. Returns the index blob and its size.

    pub(crate) fn store_checkpoint_payload(&self, state: &State) -> Result<(ObjectRef, usize)> {

        let graphs = state.checkpoint_file_graphs();

        for (_, enc) in &graphs {

            if enc.len() >= crate::state::INLINE_GRAPH_LIMIT {

                self.store_blob(enc)?;

            }

        }

        let index = state.checkpoint_index(&graphs, Repo::blob_ref);

        Ok((self.store_blob(&index)?, index.len()))

    }

    /// The graph blobs a checkpoint's payload names (none for version 1).

    pub(crate) fn checkpoint_graph_blobs(&self, cp: &Checkpoint) -> Result<Vec<ObjectRef>> {

        let payload = self.read_blob(&cp.payload)?;

        Ok(State::checkpoint_graph_refs(&payload)?)

    }

    /// Decode a checkpoint's payload, reading its graph blobs from the store.

    pub(crate) fn state_from_checkpoint(&self, payload: &[u8]) -> Result<State> {

        Ok(State::from_checkpoint(payload, |refs| {

            self.read_blobs(refs).map_err(|e| {

                crate::objects::ObjectError::Schema(format!("checkpoint graph blobs: {e}"))

            })

        })?)

    }

    /// Checkpoint the current bookmark's state now.

    pub fn checkpoint(&self) -> Result<(ObjectRef, usize, usize)> {

        let heads = self.bookmark_heads()?;

        let state = self.load_state_full(&heads)?;

        let (r, bytes) = self.write_checkpoint(&self.config.bookmark, &state, &heads)?;

        Ok((r, state.len(), bytes))

    }

    /// Point `bookmark` at an already stored checkpoint (a shallow clone's base).

    pub(crate) fn install_checkpoint(&self, bookmark: &str, r: &ObjectRef) -> Result<()> {

        Checkpoint::decode(&self.store.get_typed(r, ObjectType::Checkpoint)?)?;

        fs::create_dir_all(self.oxy.join("checkpoints"))?;

        write_atomic(&self.checkpoint_pointer(bookmark), r.to_hex().as_bytes())

    }

    /// The checkpoint of exactly `heads`: the bookmark's latest when it

    /// is that state, else one made now (kept as the bookmark's when the

    /// bookmark is at `heads`).

    fn checkpoint_ref_for(&self, heads: &[ChangeHash]) -> Result<ObjectRef> {

        let _t = crate::timing::phase("checkpoint: find or make");

        let wanted = StateObject::new(heads.to_vec()).heads;

        let bookmark = self.config.bookmark.clone();

        let current = self.bookmark_heads()?;

        Ok(match self.latest_checkpoint(&bookmark)? {

            Some((r, cp)) if cp.heads == wanted => r,

            _ => {

                let state = self.load_state(&wanted)?;

                if current == wanted {

                    self.write_checkpoint(&bookmark, &state, &wanted)?.0

                } else {

                    let (blob, _) = self.store_checkpoint_payload(&state)?;

                    let cp = Checkpoint {

                        heads: wanted.clone(),

                        count: state.len() as u64,

                        payload: blob,

                    };

                    self.store.put(ObjectType::Checkpoint, &cp.encode())?

                }

            }

        })

    }

    /// A checkpoint of exactly `heads` without its graph blobs (spec

    /// §16.1 `checkpoint-index`, oxygen#137): the checkpoint object, its

    /// payload blob and the blob's chunks. The client asks for the graph

    /// blobs the payload names in parts, with `objects`, so neither side

    /// holds a big state's whole answer at once.

    pub fn checkpoint_index_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        let r = self.checkpoint_ref_for(heads)?;

        let cp = Checkpoint::decode(&self.store.get_typed(&r, ObjectType::Checkpoint)?)?;

        let mut bundle = Bundle::default();

        let chunks = Blob::decode(&self.store.get_typed(&cp.payload, ObjectType::Blob)?)?.chunks;

        for got in self.store.get_many(&chunks) {

            bundle.objects.push(got?);

        }

        bundle.objects.push(self.store.get(&cp.payload)?);

        bundle.objects.push((ObjectType::Checkpoint, cp.encode()));

        Ok(bundle)

    }

    /// A checkpoint of exactly `heads` with its graph blobs and their

    /// chunks, in one answer (spec §16.1 `checkpoint`): what a client

    /// older than `checkpoint-index` asks for.

    pub fn checkpoint_bundle_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        let mut bundle = self.checkpoint_index_for(heads)?;

        let _t = crate::timing::phase("checkpoint: gather its objects");

        let cp = match bundle.objects.last() {

            Some((ObjectType::Checkpoint, body)) => Checkpoint::decode(body)?,

            _ => return fail("a checkpoint index without its checkpoint"),

        };

        // The graph blobs and their chunks in two parallel reads (#137): a

        // big state's are tens of thousands, and one by one they took six

        // minutes for VS Code.

        let graphs = self.checkpoint_graph_blobs(&cp)?;

        let mut blobs = Vec::with_capacity(graphs.len());

        for got in self.store.get_many(&graphs) {

            blobs.push(got?);

        }

        let mut chunks = Vec::new();

        for (ty, body) in &blobs {

            if *ty != ObjectType::Blob {

                return fail("a checkpoint graph is not a blob");

            }

            chunks.extend(Blob::decode(body)?.chunks);

        }

        let last = bundle.objects.pop();

        for got in self.store.get_many(&chunks) {

            bundle.objects.push(got?);

        }

        bundle.objects.extend(blobs);

        bundle.objects.extend(last);

        Ok(bundle)

    }

    /// Load the state of `heads` using a checkpoint when one applies.

    pub(crate) fn load_state_checkpointed(&self, heads: &[ChangeHash]) -> Result<Option<State>> {

        self.load_state_checkpointed_as(heads, false)

    }

    /// As [`Repo::load_state_checkpointed`]; `history` builds no file graph.

    pub(crate) fn load_state_checkpointed_as(

        &self,

        heads: &[ChangeHash],

        history: bool,

    ) -> Result<Option<State>> {

        let mut bookmarks: Vec<String> = vec![self.config.bookmark.clone()];

        if let Ok(rd) = fs::read_dir(self.oxy.join("checkpoints")) {

            for e in rd.flatten() {

                let name = file_name_to_name(&e.file_name().to_string_lossy());

                if !bookmarks.contains(&name) {

                    bookmarks.push(name);

                }

            }

        }

        for b in bookmarks {

            let Some((_, cp)) = self.latest_checkpoint(&b)? else {

                continue;

            };

            let payload = {

                let _t = crate::timing::phase("load: read checkpoint");

                self.read_blob(&cp.payload)?

            };

            let base = {

                let _t = crate::timing::phase("load: decode checkpoint");

                if history {

                    State::from_checkpoint_history(&payload)?

                } else {

                    self.state_from_checkpoint(&payload)?

                }

            };

            let cp_heads: BTreeSet<ChangeHash> = cp.heads.iter().copied().collect();

            let mut found: BTreeSet<ChangeHash> = BTreeSet::new();

            let mut delta: Vec<(ChangeHash, Change, Vec<crate::objects::TextRef>)> = Vec::new();

            let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();

            let mut stack: Vec<ChangeHash> = heads.to_vec();

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

                if !seen.insert(h) {

                    continue;

                }

                if base.contains(&h) {

                    if cp_heads.contains(&h) {

                        found.insert(h);

                    }

                    continue;

                }

                // For history, the skeleton: text is read only for the

                // graphs built later, and a state holding none can be

                // indexed (oxygen#93).

                let (c, texts) = if history {

                    self.store.get_change_lazy(&h)?

                } else {

                    (self.store.get_change(&h)?, Vec::new())

                };

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

                delta.push((h, c, texts));

            }

            if found.len() != cp_heads.len() {

                continue;

            }

            let _t = crate::timing::phase("load: apply delta and validate");

            let mut state = base;

            for (h, c, texts) in delta {

                if history {

                    state.add_known(h, c, texts);

                } else {

                    state.add_unchecked(c);

                }

            }

            state.validate()?;

            return Ok(Some(state));

        }

        Ok(None)

    }

}