// revlog.rs

//

// Copyright 2019-2020 Georges Racinet <georges.racinet@octobus.net>

//

// This software may be used and distributed according to the terms of the

// GNU General Public License version 2 or any later version.

use crate::{

    conversion::{rev_pyiter_collect, rev_pyiter_collect_or_else},

    utils::{node_from_py_bytes, node_from_py_object},

    PyRevision,

};

use cpython::{

    buffer::{Element, PyBuffer},

    exc::{IndexError, ValueError},

    ObjectProtocol, PyBool, PyBytes, PyClone, PyDict, PyErr, PyInt, PyList,

    PyModule, PyObject, PyResult, PySet, PyString, PyTuple, Python,

    PythonObject, ToPyObject, UnsafePyLeaked,

};

use hg::{

    errors::HgError,

    index::{

        IndexHeader, Phase, RevisionDataParams, SnapshotsCache,

        INDEX_ENTRY_SIZE,

    },

    nodemap::{Block, NodeMapError, NodeTree as CoreNodeTree},

    revlog::{nodemap::NodeMap, Graph, NodePrefix, RevlogError, RevlogIndex},

    BaseRevision, Node, Revision, UncheckedRevision, NULL_REVISION,

};

use std::{cell::RefCell, collections::HashMap};

use vcsgraph::graph::Graph as VCSGraph;

pub struct PySharedIndex {

    /// The underlying hg-core index

    pub(crate) inner: &'static hg::index::Index,

}

/// Return a Struct implementing the Graph trait

pub(crate) fn py_rust_index_to_graph(

    py: Python,

    index: PyObject,

) -> PyResult<UnsafePyLeaked<PySharedIndex>> {

    let leaked = midx.index(py).leak_immutable();

    Ok(unsafe { leaked.map(py, |idx| PySharedIndex { inner: idx }) })

}

impl Clone for PySharedIndex {

    fn clone(&self) -> Self {

        Self { inner: self.inner }

    }

}

impl Graph for PySharedIndex {

    fn parents(&self, rev: Revision) -> Result<[Revision; 2], hg::GraphError> {

        self.inner.parents(rev)

    }

}

impl VCSGraph for PySharedIndex {

    fn parents(

        &self,

        rev: BaseRevision,

    ) -> Result<vcsgraph::graph::Parents, vcsgraph::graph::GraphReadError>

    {

        // FIXME This trait should be reworked to decide between Revision

        // and UncheckedRevision, get better errors names, etc.

        match Graph::parents(self, Revision(rev)) {

            Ok(parents) => {

                Ok(vcsgraph::graph::Parents([parents[0].0, parents[1].0]))

            }

            Err(hg::GraphError::ParentOutOfRange(rev)) => {

                Err(vcsgraph::graph::GraphReadError::KeyedInvalidKey(rev.0))

            }

        }

    }

}

impl RevlogIndex for PySharedIndex {

    fn len(&self) -> usize {

        self.inner.len()

    }

    fn node(&self, rev: Revision) -> Option<&Node> {

        self.inner.node(rev)

    }

}

    @shared data index: hg::index::Index;

    data nt: RefCell<Option<CoreNodeTree>>;

    data docket: RefCell<Option<PyObject>>;

    // Holds a reference to the mmap'ed persistent nodemap data

    data nodemap_mmap: RefCell<Option<PyBuffer>>;

    // Holds a reference to the mmap'ed persistent index data

    data index_mmap: RefCell<Option<PyBuffer>>;

    def __new__(

        _cls,

        data: PyObject,

        default_header: u32,

        Self::new(py, data, default_header)

    }

    /// Compatibility layer used for Python consumers needing access to the C index

    ///

    /// Only use case so far is `scmutil.shortesthexnodeidprefix`,

    /// that may need to build a custom `nodetree`, based on a specified revset.

    /// With a Rust implementation of the nodemap, we will be able to get rid of

    /// this, by exposing our own standalone nodemap class,

    /// ready to accept `Index`.

/*    def get_cindex(&self) -> PyResult<PyObject> {

        Ok(self.cindex(py).borrow().inner().clone_ref(py))

    }

*/

    // Index API involving nodemap, as defined in mercurial/pure/parsers.py

    /// Return Revision if found, raises a bare `error.RevlogError`

    /// in case of ambiguity, same as C version does

    def get_rev(&self, node: PyBytes) -> PyResult<Option<PyRevision>> {

        let opt = self.get_nodetree(py)?.borrow();

        let nt = opt.as_ref().unwrap();

        let ridx = &*self.index(py).borrow();

        let node = node_from_py_bytes(py, &node)?;

        let rust_rev =

            nt.find_bin(ridx, node.into()).map_err(|e| nodemap_error(py, e))?;

        Ok(rust_rev.map(Into::into))

    }

    /// same as `get_rev()` but raises a bare `error.RevlogError` if node

    /// is not found.

    ///

    /// No need to repeat `node` in the exception, `mercurial/revlog.py`

    /// will catch and rewrap with it

    def rev(&self, node: PyBytes) -> PyResult<PyRevision> {

        self.get_rev(py, node)?.ok_or_else(|| revlog_error(py))

    }

    /// return True if the node exist in the index

    def has_node(&self, node: PyBytes) -> PyResult<bool> {

        // TODO OPTIM we could avoid a needless conversion here,

        // to do when scaffolding for pure Rust switch is removed,

        // as `get_rev()` currently does the necessary assertions

        self.get_rev(py, node).map(|opt| opt.is_some())

    }

    /// find length of shortest hex nodeid of a binary ID

    def shortest(&self, node: PyBytes) -> PyResult<usize> {

        let opt = self.get_nodetree(py)?.borrow();

        let nt = opt.as_ref().unwrap();

        let idx = &*self.index(py).borrow();

        match nt.unique_prefix_len_node(idx, &node_from_py_bytes(py, &node)?)

        {

            Ok(Some(l)) => Ok(l),

            Ok(None) => Err(revlog_error(py)),

            Err(e) => Err(nodemap_error(py, e)),

        }

    }

    def partialmatch(&self, node: PyObject) -> PyResult<Option<PyBytes>> {

        let opt = self.get_nodetree(py)?.borrow();

        let nt = opt.as_ref().unwrap();

        let idx = &*self.index(py).borrow();

        let node_as_string = if cfg!(feature = "python3-sys") {

            node.cast_as::<PyString>(py)?.to_string(py)?.to_string()

        }

        else {

            let node = node.extract::<PyBytes>(py)?;

            String::from_utf8_lossy(node.data(py)).to_string()

        };

        let prefix = NodePrefix::from_hex(&node_as_string)

            .map_err(|_| PyErr::new::<ValueError, _>(

                py, format!("Invalid node or prefix '{}'", node_as_string))

            )?;

        nt.find_bin(idx, prefix)

            // TODO make an inner API returning the node directly

            .map(|opt| opt.map(

                |rev| PyBytes::new(py, idx.node(rev).unwrap().as_bytes())))

            .map_err(|e| nodemap_error(py, e))

    }

    /// append an index entry

    def append(&self, tup: PyTuple) -> PyResult<PyObject> {

        if tup.len(py) < 8 {

            // this is better than the panic promised by tup.get_item()

            return Err(

                PyErr::new::<IndexError, _>(py, "tuple index out of range"))

        }

        let node_bytes = tup.get_item(py, 7).extract(py)?;

        let node = node_from_py_object(py, &node_bytes)?;

        let rev = self.len(py)? as BaseRevision;

        // This is ok since we will just add the revision to the index

        let rev = Revision(rev);

        self.index(py)

            .borrow_mut()

            .append(py_tuple_to_revision_data_params(py, tup)?)

            .unwrap();

        let idx = &*self.index(py).borrow();

        self.get_nodetree(py)?.borrow_mut().as_mut().unwrap()

            .insert(idx, &node, rev)

            .map_err(|e| nodemap_error(py, e))?;

        Ok(py.None())

    }

    def __delitem__(&self, key: PyObject) -> PyResult<()> {

        // __delitem__ is both for `del idx[r]` and `del idx[r1:r2]`

        let start = if let Ok(rev) = key.extract(py) {

            UncheckedRevision(rev)

        } else {

            let start = key.getattr(py, "start")?;

            UncheckedRevision(start.extract(py)?)

        };

        let start = self.index(py)

            .borrow()

            .check_revision(start)

            .ok_or_else(|| {

                nodemap_error(py, NodeMapError::RevisionNotInIndex(start))

            })?;

        self.index(py).borrow_mut().remove(start).unwrap();

        let mut opt = self.get_nodetree(py)?.borrow_mut();

        let nt = opt.as_mut().unwrap();

        nt.invalidate_all();

        self.fill_nodemap(py, nt)?;

        Ok(())

    }

    //

    // Index methods previously reforwarded to C index (tp_methods)

    // Same ordering as in revlog.c

    //

    /// return the gca set of the given revs

    def ancestors(&self, *args, **_kw) -> PyResult<PyObject> {

        let rust_res = self.inner_ancestors(py, args)?;

        Ok(rust_res)

    }

    /// return the heads of the common ancestors of the given revs

    def commonancestorsheads(&self, *args, **_kw) -> PyResult<PyObject> {

        let rust_res = self.inner_commonancestorsheads(py, args)?;

        Ok(rust_res)

    }

    /// Clear the index caches and inner py_class data.

    /// It is Python's responsibility to call `update_nodemap_data` again.

    def clearcaches(&self) -> PyResult<PyObject> {

        self.nt(py).borrow_mut().take();

        self.docket(py).borrow_mut().take();

        self.nodemap_mmap(py).borrow_mut().take();

        self.index(py).borrow().clear_caches();

        Ok(py.None())

    }

    /// return the raw binary string representing a revision

    def entry_binary(&self, *args, **_kw) -> PyResult<PyObject> {

        let rindex = self.index(py).borrow();

        let rev = UncheckedRevision(args.get_item(py, 0).extract(py)?);

        let rust_bytes = rindex.check_revision(rev).and_then(

            |r| rindex.entry_binary(r))

            .ok_or_else(|| rev_not_in_index(py, rev))?;

        let rust_res = PyBytes::new(py, rust_bytes).into_object();

        Ok(rust_res)

    }

    /// return a binary packed version of the header

    def pack_header(&self, *args, **_kw) -> PyResult<PyObject> {

        let rindex = self.index(py).borrow();

        let packed = rindex.pack_header(args.get_item(py, 0).extract(py)?);

        let rust_res = PyBytes::new(py, &packed).into_object();

        Ok(rust_res)

    }

    /// compute phases

    def computephasesmapsets(&self, *args, **_kw) -> PyResult<PyObject> {

        let py_roots = args.get_item(py, 0).extract::<PyDict>(py)?;

        let rust_res = self.inner_computephasesmapsets(py, py_roots)?;

        Ok(rust_res)

    }

    /// reachableroots

    def reachableroots2(&self, *args, **_kw) -> PyResult<PyObject> {

        let rust_res = self.inner_reachableroots2(

            py,

            UncheckedRevision(args.get_item(py, 0).extract(py)?),

            args.get_item(py, 1),

            args.get_item(py, 2),

            args.get_item(py, 3).extract(py)?,

        )?;

        Ok(rust_res)

    }

    /// get head revisions

    def headrevs(&self) -> PyResult<PyObject> {

        let rust_res = self.inner_headrevs(py)?;

        Ok(rust_res)

    }

    /// get filtered head revisions

    def headrevsfiltered(&self, *args, **_kw) -> PyResult<PyObject> {

        let rust_res = self.inner_headrevsfiltered(py, &args.get_item(py, 0))?;

        Ok(rust_res)

    }

    /// True if the object is a snapshot

    def issnapshot(&self, *args, **_kw) -> PyResult<bool> {

        let index = self.index(py).borrow();

        let result = index

            .is_snapshot(UncheckedRevision(args.get_item(py, 0).extract(py)?))

            .map_err(|e| {

                PyErr::new::<cpython::exc::ValueError, _>(py, e.to_string())

            })?;

        Ok(result)

    }

    /// Gather snapshot data in a cache dict

    def findsnapshots(&self, *args, **_kw) -> PyResult<PyObject> {

        let index = self.index(py).borrow();

        let cache: PyDict = args.get_item(py, 0).extract(py)?;

        // this methods operates by setting new values in the cache,

        // hence we will compare results by letting the C implementation

        // operate over a deepcopy of the cache, and finally compare both

        // caches.

        let c_cache = PyDict::new(py);

        for (k, v) in cache.items(py) {

            c_cache.set_item(py, k, PySet::new(py, v)?)?;

        }

        let start_rev = UncheckedRevision(args.get_item(py, 1).extract(py)?);

        let end_rev = UncheckedRevision(args.get_item(py, 2).extract(py)?);

        let mut cache_wrapper = PySnapshotsCache{ py, dict: cache };

        index.find_snapshots(

            start_rev,

            end_rev,

            &mut cache_wrapper,

        ).map_err(|_| revlog_error(py))?;

        Ok(py.None())

    }

    /// determine revisions with deltas to reconstruct fulltext

    def deltachain(&self, *args, **_kw) -> PyResult<PyObject> {

        let index = self.index(py).borrow();

        let rev = args.get_item(py, 0).extract::<BaseRevision>(py)?.into();

        let stop_rev =

            args.get_item(py, 1).extract::<Option<BaseRevision>>(py)?;

        let rev = index.check_revision(rev).ok_or_else(|| {

            nodemap_error(py, NodeMapError::RevisionNotInIndex(rev))

        })?;

        let stop_rev = if let Some(stop_rev) = stop_rev {

            let stop_rev = UncheckedRevision(stop_rev);

            Some(index.check_revision(stop_rev).ok_or_else(|| {

                nodemap_error(py, NodeMapError::RevisionNotInIndex(stop_rev))

            })?)

        } else {None};

        let using_general_delta = args.get_item(py, 2)

            .extract::<Option<u32>>(py)?

            .map(|i| i != 0);

        let (chain, stopped) = index.delta_chain(

            rev, stop_rev, using_general_delta

        ).map_err(|e| {

            PyErr::new::<cpython::exc::ValueError, _>(py, e.to_string())

        })?;

        let chain: Vec<_> = chain.into_iter().map(|r| r.0).collect();

        Ok(

            PyTuple::new(

                py,

                &[

                    chain.into_py_object(py).into_object(),

                    stopped.into_py_object(py).into_object()

                ]

            ).into_object()

        )

    }

    /// slice planned chunk read to reach a density threshold

    def slicechunktodensity(&self, *args, **_kw) -> PyResult<PyObject> {

        let rust_res = self.inner_slicechunktodensity(

            py,

            args.get_item(py, 0),

            args.get_item(py, 1).extract(py)?,

            args.get_item(py, 2).extract(py)?

        )?;

        Ok(rust_res)

    }

    // index_sequence_methods and index_mapping_methods.

    //

    // Since we call back through the high level Python API,

    // there's no point making a distinction between index_get

    // and index_getitem.

    // gracinet 2023: this above is no longer true for the pure Rust impl

    def __len__(&self) -> PyResult<usize> {

        self.len(py)

    }

    def __getitem__(&self, key: PyObject) -> PyResult<PyObject> {

        let rust_res = self.inner_getitem(py, key.clone_ref(py))?;

        Ok(rust_res)

    }

    def __contains__(&self, item: PyObject) -> PyResult<bool> {

        // ObjectProtocol does not seem to provide contains(), so

        // this is an equivalent implementation of the index_contains()

        // defined in revlog.c

        match item.extract::<i32>(py) {

            Ok(rev) => {

                Ok(rev >= -1 && rev < self.len(py)? as BaseRevision)

            }

            Err(_) => {

                let item_bytes: PyBytes = item.extract(py)?;

                let rust_res = self.has_node(py, item_bytes)?;

                Ok(rust_res)

            }

        }

    }

    def nodemap_data_all(&self) -> PyResult<PyBytes> {

        self.inner_nodemap_data_all(py)

    }

    def nodemap_data_incremental(&self) -> PyResult<PyObject> {

        self.inner_nodemap_data_incremental(py)

    }

    def update_nodemap_data(

        &self,

        docket: PyObject,

        nm_data: PyObject

    ) -> PyResult<PyObject> {

        self.inner_update_nodemap_data(py, docket, nm_data)

    }

    @property

    def entry_size(&self) -> PyResult<PyInt> {

        let rust_res: PyInt = INDEX_ENTRY_SIZE.to_py_object(py);

        Ok(rust_res)

    }

    @property

    def rust_ext_compat(&self) -> PyResult<PyInt> {

        // will be entirely removed when the Rust index yet useful to

        // implement in Rust to detangle things when removing `self.cindex`

        let rust_res: PyInt = 1.to_py_object(py);

        Ok(rust_res)

    }

    @property

    def is_rust(&self) -> PyResult<PyBool> {

        Ok(false.to_py_object(py))

    }

});

/// Take a (potentially) mmap'ed buffer, and return the underlying Python

/// buffer along with the Rust slice into said buffer. We need to keep the

/// Python buffer around, otherwise we'd get a dangling pointer once the buffer

/// is freed from Python's side.

///

/// # Safety

///

/// The caller must make sure that the buffer is kept around for at least as

/// long as the slice.

#[deny(unsafe_op_in_unsafe_fn)]

unsafe fn mmap_keeparound(

    py: Python,

    data: PyObject,

) -> PyResult<(

    PyBuffer,

    Box<dyn std::ops::Deref<Target = [u8]> + Send + Sync + 'static>,

)> {

    let buf = PyBuffer::get(py, &data)?;

    let len = buf.item_count();

    // Build a slice from the mmap'ed buffer data

    let cbuf = buf.buf_ptr();

    let bytes = if std::mem::size_of::<u8>() == buf.item_size()

        && buf.is_c_contiguous()