// revlog.rs

//

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

//           2020-2024 Raphaël Gomès <raphael.gomes@octobus.net>

//           2024 Georges Racinet <georges.racinet@cloudcrane.io>

//

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

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

#![allow(non_snake_case)]

use hg::revlog::index::IndexHeader;

use hg::revlog::nodemap::Block;

use hg::utils::files::get_bytes_from_path;

use pyo3::buffer::PyBuffer;

use pyo3::conversion::IntoPyObject;

use pyo3::exceptions::{PyIndexError, PyTypeError, PyValueError};

use pyo3::types::{

    PyBool, PyBytes, PyBytesMethods, PyDict, PyList, PySet, PyTuple,

};

use pyo3::{prelude::*, IntoPyObjectExt};

use pyo3_sharedref::{PyShareable, SharedByPyObject};

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

use std::os::fd::AsRawFd;

use std::sync::{

    atomic::{AtomicUsize, Ordering},

    RwLock, RwLockReadGuard, RwLockWriteGuard,

};

use hg::{

    errors::HgError,

    revlog::{

        index::{

            Index, Phase, RevisionDataParams, SnapshotsCache, INDEX_ENTRY_SIZE,

        },

        inner_revlog::InnerRevlog as CoreInnerRevlog,

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

        options::RevlogOpenOptions,

        RevlogError, RevlogIndex, RevlogType,

    },

    utils::files::get_path_from_bytes,

    vfs::FnCacheVfs,

    BaseRevision, Revision, UncheckedRevision, NULL_REVISION,

};

use crate::utils::PyBytesDeref;

use crate::{

    exceptions::{

        graph_error, map_lock_error, map_try_lock_error, nodemap_error,

        rev_not_in_index, revlog_error_bare, revlog_error_from_msg,

    },

    node::{node_from_py_bytes, node_prefix_from_py_bytes, py_node_for_rev},

    revision::{

        check_revision, rev_pyiter_collect, rev_pyiter_collect_or_else,

        revs_py_list, revs_py_set, PyRevision,

    },

    store::PyFnCache,

    transaction::PyTransaction,

    utils::{new_submodule, take_buffer_with_slice, with_pybytes_buffer},

};

mod config;

use config::*;

mod index;

pub use index::PySharedIndex;

use index::{

    py_tuple_to_revision_data_params, revision_data_params_to_py_tuple,

};

#[pyclass]

struct ReadingContextManager {

    inner_revlog: Py<InnerRevlog>,

}

#[pymethods]

impl ReadingContextManager {

    fn __enter__(slf: PyRef<'_, Self>) -> PyResult<()> {

        let inner_bound = slf.inner_revlog.bind(slf.py());

        let shareable = &inner_bound.borrow().irl;

        // Safety: the owner is correct and we won't use `share()` anyway

        let core_irl =

            unsafe { shareable.borrow_with_owner(inner_bound) }.read();

        core_irl

            .enter_reading_context()

            .map_err(revlog_error_from_msg)

            .inspect_err(|_e| {

                // `__exit__` is not called from Python if `__enter__` fails

                core_irl.exit_reading_context();

            })

    }

    #[pyo3(signature = (*_args))]

    fn __exit__(slf: PyRef<'_, Self>, _args: &Bound<'_, PyTuple>) {

        let inner_bound = slf.inner_revlog.bind(slf.py());

        let shareable = &inner_bound.borrow().irl;

        // Safety: the owner is correct and we won't use `share()` anyway

        let core_irl_ref = unsafe { shareable.borrow_with_owner(inner_bound) };

        core_irl_ref.read().exit_reading_context();

    }

}

#[pyclass]

struct WritingContextManager {

    inner_revlog: Py<InnerRevlog>,

    transaction: RwLock<PyTransaction>,

    data_end: Option<usize>,

}

#[pymethods]

impl WritingContextManager {

    fn __enter__(slf: PyRefMut<'_, Self>) -> PyResult<()> {

        let inner_bound = slf.inner_revlog.bind(slf.py());

        let shareable = &inner_bound.borrow_mut().irl;

        // Safety: the owner is correct and we won't use `share()` anyway

        let mut core_irl =

            unsafe { shareable.borrow_with_owner(inner_bound) }.write();

        core_irl

            .enter_writing_context(

                slf.data_end,

                &mut *slf

                    .transaction

                    .try_write()

                    .expect("transaction should be protected by the GIL"),

            )

            .map_err(revlog_error_from_msg)

            .inspect_err(|_e| {

                // `__exit__` is not called from Python if `__enter__` fails

                core_irl.exit_writing_context();

            })

    }

    #[pyo3(signature = (*_args))]

    fn __exit__(slf: PyRef<'_, Self>, _args: &Bound<'_, PyTuple>) {

        let inner_bound = slf.inner_revlog.bind(slf.py());

        let shareable = &inner_bound.borrow().irl;

        // Safety: the owner is correct and we won't use `share()` anyway

        let core_irl_ref = unsafe { shareable.borrow_with_owner(inner_bound) };

        core_irl_ref.write().exit_writing_context();

    }

}

struct PySnapshotsCache<'a, 'py: 'a>(&'a Bound<'py, PyDict>);

impl<'a, 'py> PySnapshotsCache<'a, 'py> {

    fn insert_for_with_py_result(

        &self,

        rev: BaseRevision,

        value: BaseRevision,

    ) -> PyResult<()> {

        match self.0.get_item(rev)? {

            Some(obj) => obj.downcast::<PySet>()?.add(value),

            None => {

                let set = PySet::new(self.0.py(), vec![value])?;

                self.0.set_item(rev, set)

            }

        }

    }

}

impl<'a, 'py> SnapshotsCache for PySnapshotsCache<'a, 'py> {

    fn insert_for(

        &mut self,

        rev: BaseRevision,

        value: BaseRevision,

    ) -> Result<(), RevlogError> {

        self.insert_for_with_py_result(rev, value).map_err(|_| {

            RevlogError::Other(HgError::unsupported(

                "Error in Python caches handling",

            ))

        })

    }

}

// Only used from Python *tests*

#[doc(hidden)]

#[pyclass]

pub struct PyFileHandle {

    inner_file: std::os::fd::RawFd,

}

#[pymethods]

impl PyFileHandle {

    #[new]

    fn new(handle: std::os::fd::RawFd) -> Self {

        Self { inner_file: handle }

    }

    fn tell(&self, py: Python<'_>) -> PyResult<PyObject> {

        let locals = PyDict::new(py);

        locals.set_item("os", py.import("os")?)?;

        locals.set_item("fd", self.inner_file)?;

        let f = py.eval(c"os.fdopen(fd)", None, Some(&locals))?;

        // Prevent Python from closing the file after garbage collecting.

        // This is fine since Rust is still holding on to the actual File.

        // (and also because it's only used in tests).

        std::mem::forget(f.clone());

        locals.set_item("f", f)?;

        let res = py.eval(c"f.tell()", None, Some(&locals))?;

        Ok(res.unbind())

    }

}

#[pyclass]

#[allow(dead_code)]

pub(crate) struct InnerRevlog {

    pub(crate) irl: PyShareable<CoreInnerRevlog>,

    nt: RwLock<Option<CoreNodeTree>>,

    docket: Option<PyObject>,

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

    nodemap_mmap: Option<PyBuffer<u8>>,

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

    index_mmap: Option<PyBuffer<u8>>,

    revision_cache: Option<PyObject>,

    head_revs_py_list: Option<Py<PyList>>,

    head_node_ids_py_list: Option<Py<PyList>>,

    use_persistent_nodemap: bool,

    nodemap_queries: AtomicUsize,

}

#[pymethods]

impl InnerRevlog {

    #[new]

    // The Python side has authority on this signature.

    #[allow(clippy::too_many_arguments)]

    fn new(

        vfs_base: &Bound<'_, PyBytes>,

        fncache: &Bound<'_, PyAny>,

        vfs_is_readonly: bool,

        index_data: &Bound<'_, PyAny>,

        index_file: &Bound<'_, PyBytes>,

        data_file: &Bound<'_, PyBytes>,

        sidedata_file: &Bound<'_, PyAny>,

        inline: bool,

        data_config: &Bound<'_, PyAny>,

        delta_config: &Bound<'_, PyAny>,

        feature_config: &Bound<'_, PyAny>,

        chunk_cache: &Bound<'_, PyAny>,

        default_compression_header: &Bound<'_, PyAny>,

        revlog_type: usize,

        use_persistent_nodemap: bool,

    ) -> PyResult<Self> {

        // Let clippy accept the unused arguments. This is a bit better than

        // a blank `allow` directive

        let _ = sidedata_file;

        let _ = chunk_cache;

        let _ = default_compression_header;

        let index_file = get_path_from_bytes(index_file.as_bytes()).to_owned();

        let data_file = get_path_from_bytes(data_file.as_bytes()).to_owned();

        let revlog_type = RevlogType::try_from(revlog_type)

            .map_err(revlog_error_from_msg)?;

        let data_config = extract_data_config(data_config, revlog_type)?;

        let delta_config = extract_delta_config(delta_config, revlog_type)?;

        let feature_config =

            extract_feature_config(feature_config, revlog_type)?;

        let options = RevlogOpenOptions::new(

            inline,

            data_config,

            delta_config,

            feature_config,

        );

        // Safety: we keep the buffer around inside the returned instance as

        // `index_mmap`

        let (buf, bytes) = unsafe { take_buffer_with_slice(index_data)? };

        let index = Index::new(bytes, options.index_header())

            .map_err(revlog_error_from_msg)?;

        let base = get_path_from_bytes(vfs_base.as_bytes()).to_owned();

        let core = CoreInnerRevlog::new(

            Box::new(FnCacheVfs::new(

                base,

                vfs_is_readonly,

                Box::new(PyFnCache::new(fncache.clone().unbind())),

            )),

            index,

            index_file,

            data_file,

            data_config,

            delta_config,

            feature_config,

        );

        Ok(Self {

            irl: core.into(),

            nt: None.into(),

            docket: None,

            nodemap_mmap: None,

            index_mmap: buf.into(),

            head_revs_py_list: None,

            head_node_ids_py_list: None,

            revision_cache: None,

            use_persistent_nodemap,

            nodemap_queries: AtomicUsize::new(0),

        })

    }

    #[getter]

    fn canonical_index_file(

        slf: &Bound<'_, Self>,

        py: Python<'_>,

    ) -> PyResult<Py<PyBytes>> {

        Self::with_core_read(slf, |_self_ref, irl| {

            let path = irl.canonical_index_file();

            Ok(PyBytes::new(py, &get_bytes_from_path(path)).into())

        })

    }

    #[getter]

    fn is_delaying(slf: &Bound<'_, Self>) -> PyResult<bool> {

        Self::with_core_read(slf, |_self_ref, irl| Ok(irl.is_delaying()))

    }

    #[getter]

    fn inline(slf: &Bound<'_, Self>) -> PyResult<bool> {

        Self::with_core_read(slf, |_self_ref, irl| Ok(irl.is_inline()))

    }

    #[setter]

    fn set_inline(slf: &Bound<'_, Self>, inline: bool) -> PyResult<()> {

        Self::with_core_write(slf, |_self_ref, mut irl| {

            irl.inline = inline;

            Ok(())

        })

    }

    #[getter]

    fn is_writing(slf: &Bound<'_, Self>) -> PyResult<bool> {

        Self::with_core_read(slf, |_self_ref, irl| Ok(irl.is_writing()))

    }

    #[getter]

    fn is_open(slf: &Bound<'_, Self>) -> PyResult<bool> {

        Self::with_core_read(slf, |_self_ref, irl| Ok(irl.is_open()))

    }

    #[getter]

    fn _revisioncache(&self, py: Python<'_>) -> PyResult<PyObject> {

        match &self.revision_cache {

            None => Ok(py.None()),

            Some(cache) => Ok(cache.clone_ref(py)),

        }

    }

    #[setter]

    fn set__revisioncache(

        slf: &Bound<'_, Self>,

        py: Python<'_>,

        value: Option<PyObject>,

    ) -> PyResult<()> {

        let mut self_ref = slf.borrow_mut();

        self_ref.revision_cache = value.as_ref().map(|v| v.clone_ref(py));

        match value {

            None => {

                // This means the property has been deleted, *not* that the

                // property has been set to `None`. Whatever happens is up

                // to the implementation. Here we just set it to `None`.

                self_ref.revision_cache.take();

            }

            Some(tuple) => {

                if tuple.is_none(py) {

                    self_ref.revision_cache.take();

                    return Ok(());

                }

                drop(self_ref);

                let tuple: &Bound<'_, PyTuple> = tuple.downcast_bound(py)?;

                let node = tuple.get_item(0)?;

                let node = node_from_py_bytes(node.downcast()?)?;

                let rev: BaseRevision = tuple.get_item(1)?.extract()?;

                // Ok because Python only sets this if the revision has been

                // checked

                let rev = Revision(rev);

                let data = tuple.get_item(2)?;

                let bytes = data.downcast_into::<PyBytes>()?.unbind();

                Self::with_core_read(slf, |_self_ref, irl| {

                    let mut last_revision_cache = irl

                        .last_revision_cache

                        .lock()

                        .expect("lock should not be held");

                    *last_revision_cache = Some((

                        node,

                        rev,

                        Box::new(PyBytesDeref::new(py, bytes)),

                    ));

                    Ok(())

                })?;

            }

        }

        Ok(())

    }

    #[getter]

    fn index_file(

        slf: &Bound<'_, Self>,

        py: Python<'_>,

    ) -> PyResult<Py<PyBytes>> {

        Self::with_core_read(slf, |_self_ref, irl| {

            let path = get_bytes_from_path(&irl.index_file);

            Ok(PyBytes::new(py, &path).unbind())

        })

    }

    #[setter]

    fn set_index_file(

        slf: &Bound<'_, Self>,

        path: &Bound<'_, PyBytes>,

    ) -> PyResult<()> {

        Self::with_core_write(slf, |_self_ref, mut irl| {

            let path = get_path_from_bytes(path.as_bytes());

            path.clone_into(&mut irl.index_file);

            Ok(())

        })

    }

    // This is only used in Python *tests*

    #[getter]

    #[doc(hidden)]

    fn _writinghandles(

        slf: &Bound<'_, Self>,

        py: Python<'_>,

    ) -> PyResult<PyObject> {

        Self::with_core_read(slf, |_self_ref, irl| {

            let handles = irl.python_writing_handles();

            match handles.as_ref() {

                None => Ok(py.None()),

                Some(handles) => {

                    let index_handle = PyFileHandle::new(

                        handles.index_handle.file.as_raw_fd(),

                    );

                    let data_handle = handles

                        .data_handle

                        .as_ref()

                        .map(|h| PyFileHandle::new(h.file.as_raw_fd()));

                    Ok(PyTuple::new(

                        py,

                        &[

                            index_handle.into_py_any(py)?,

                            data_handle.into_py_any(py)?,

                            py.None(), // Sidedata handle

                        ],

                    )?

                    .unbind()

                    .into())

                }

            }

        })

    }

    fn clear_cache(slf: &Bound<'_, Self>) -> PyResult<PyObject> {

        assert!(!Self::is_delaying(slf)?);

        let mut self_ref = slf.borrow_mut();

        self_ref.revision_cache.take();

        self_ref.nodemap_queries.store(0, Ordering::Relaxed);

        drop(self_ref);

        Self::with_core_write(slf, |_self_ref, mut irl| {

            irl.clear_cache();

            Ok(slf.py().None())

        })

    }

    fn issnapshot(slf: &Bound<'_, Self>, rev: PyRevision) -> PyResult<bool> {

        Self::_index_issnapshot(slf, rev)

    }

    #[pyo3(signature = (rev, stoprev=None))]

    fn _deltachain(

        slf: &Bound<'_, Self>,

        py: Python<'_>,

        rev: PyRevision,

        stoprev: Option<PyRevision>,

    ) -> PyResult<Py<PyTuple>> {

    }

    fn compress(

        slf: &Bound<'_, Self>,

        py: Python<'_>,

        data: &Bound<'_, PyAny>,

    ) -> PyResult<Py<PyTuple>> {

        Self::with_core_read(slf, |_self_ref, irl| {

            // Safety: we only hold on to the data for as long as `_buf`

            // is alive

            let (_buf, data) = unsafe { take_buffer_with_slice(data)? };

            let compressed =

                irl.compress(&data).map_err(revlog_error_from_msg)?;

            let compressed = compressed.as_deref();

            let header = if compressed.is_some() {

                PyBytes::new(py, &b""[..])

            } else {

                PyBytes::new(py, &b"u"[..])