//! 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(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())
    }

    /// A bundle holding a checkpoint of exactly `heads` with its payload
    /// (spec §16.1 `checkpoint`); the current bookmark checkpoint is reused
    /// when it matches, otherwise one is made and, when `heads` are the
    /// bookmark's, kept.
    pub fn checkpoint_bundle_for(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        let wanted = StateObject::new(heads.to_vec()).heads;
        let bookmark = self.config.bookmark.clone();
        let current = self.bookmark_heads()?;
        let r = 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())?
                }
            }
        };
        let cp = Checkpoint::decode(&self.store.get_typed(&r, ObjectType::Checkpoint)?)?;
        let mut bundle = Bundle::default();
        for c in Blob::decode(&self.store.get_typed(&cp.payload, ObjectType::Blob)?)?.chunks {
            bundle.objects.push(self.store.get(&c)?);
        }
        bundle.objects.push(self.store.get(&cp.payload)?);
        for g in self.checkpoint_graph_blobs(&cp)? {
            for c in Blob::decode(&self.store.get_typed(&g, ObjectType::Blob)?)?.chunks {
                bundle.objects.push(self.store.get(&c)?);
            }
            bundle.objects.push(self.store.get(&g)?);
        }
        bundle.objects.push((ObjectType::Checkpoint, cp.encode()));
        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>> {
        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 = e.file_name().to_string_lossy().to_string();
                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");
                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<Change> = 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;
                }
                let c = self.store.get_change(&h)?;
                stack.extend(c.deps.iter().copied());
                delta.push(c);
            }
            if found.len() != cp_heads.len() {
                continue;
            }
            let _t = crate::timing::phase("load: apply delta and validate");
            let mut state = base;
            for c in delta {
                state.add_unchecked(c);
            }
            state.validate()?;
            return Ok(Some(state));
        }
        Ok(None)
    }
}