# parsers.py - Python implementation of parsers.c

#

# Copyright 2009 Olivia Mackall <olivia@selenic.com> and others

#

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

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

from __future__ import absolute_import

import struct

import zlib

from ..node import (

    nullrev,

    sha1nodeconstants,

)

from .. import (

    error,

    pycompat,

    revlogutils,

    util,

)

from ..revlogutils import nodemap as nodemaputil

from ..revlogutils import constants as revlog_constants

stringio = pycompat.bytesio

_pack = struct.pack

_unpack = struct.unpack

_compress = zlib.compress

_decompress = zlib.decompress

class dirstatetuple(object):

    """represent a dirstate entry

    It contains:

    - state (one of 'n', 'a', 'r', 'm')

    - mode,

    - size,

    - mtime,

    """

    __slot__ = ('_state', '_mode', '_size', '_mtime')

    def __init__(self, state, mode, size, mtime):

        self._state = state

        self._mode = mode

        self._size = size

        self._mtime = mtime

    def __getitem__(self, idx):

        if idx == 0 or idx == -4:

            return self._state

        elif idx == 1 or idx == -3:

            return self._mode

        elif idx == 2 or idx == -2:

            return self._size

        elif idx == 3 or idx == -1:

            return self._mtime

        else:

            raise IndexError(idx)

    @property

    def state(self):

        """

        States are:

          n  normal

          m  needs merging

          r  marked for removal

          a  marked for addition

        XXX This "state" is a bit obscure and mostly a direct expression of the

        dirstatev1 format. It would make sense to ultimately deprecate it in

        favor of the more "semantic" attributes.

        """

        return self._state

    def v1_state(self):

        """return a "state" suitable for v1 serialization"""

        return self._state

    def v1_mode(self):

        """return a "mode" suitable for v1 serialization"""

        return self._mode

    def v1_size(self):

        """return a "size" suitable for v1 serialization"""

        return self._size

    def v1_mtime(self):

        """return a "mtime" suitable for v1 serialization"""

        return self._mtime

def gettype(q):

    return int(q & 0xFFFF)

class BaseIndexObject(object):

    # Can I be passed to an algorithme implemented in Rust ?

    rust_ext_compat = 0

    # Format of an index entry according to Python's `struct` language

    index_format = revlog_constants.INDEX_ENTRY_V1

    # Size of a C unsigned long long int, platform independent

    big_int_size = struct.calcsize(b'>Q')

    # Size of a C long int, platform independent

    int_size = struct.calcsize(b'>i')

    # An empty index entry, used as a default value to be overridden, or nullrev

    null_item = (

        0,

        0,

        0,

        -1,

        -1,

        -1,

        -1,

        sha1nodeconstants.nullid,

        0,

        0,

        revlog_constants.COMP_MODE_INLINE,

        revlog_constants.COMP_MODE_INLINE,

    )

    @util.propertycache

    def entry_size(self):

        return self.index_format.size

    @property

    def nodemap(self):

        msg = b"index.nodemap is deprecated, use index.[has_node|rev|get_rev]"

        util.nouideprecwarn(msg, b'5.3', stacklevel=2)

        return self._nodemap

    @util.propertycache

    def _nodemap(self):

        nodemap = nodemaputil.NodeMap({sha1nodeconstants.nullid: nullrev})

        for r in range(0, len(self)):

            n = self[r][7]

            nodemap[n] = r

        return nodemap

    def has_node(self, node):

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

        return node in self._nodemap

    def rev(self, node):

        """return a revision for a node

        If the node is unknown, raise a RevlogError"""

        return self._nodemap[node]

    def get_rev(self, node):

        """return a revision for a node

        If the node is unknown, return None"""

        return self._nodemap.get(node)

    def _stripnodes(self, start):

        if '_nodemap' in vars(self):

            for r in range(start, len(self)):

                n = self[r][7]

                del self._nodemap[n]

    def clearcaches(self):

        self.__dict__.pop('_nodemap', None)

    def __len__(self):

        return self._lgt + len(self._extra)

    def append(self, tup):

        if '_nodemap' in vars(self):

            self._nodemap[tup[7]] = len(self)

        data = self._pack_entry(len(self), tup)

        self._extra.append(data)

    def _pack_entry(self, rev, entry):

        assert entry[8] == 0

        assert entry[9] == 0

        return self.index_format.pack(*entry[:8])

    def _check_index(self, i):

        if not isinstance(i, int):

            raise TypeError(b"expecting int indexes")

        if i < 0 or i >= len(self):

            raise IndexError

    def __getitem__(self, i):

        if i == -1:

            return self.null_item

        self._check_index(i)

        if i >= self._lgt:

            data = self._extra[i - self._lgt]

        else:

            index = self._calculate_index(i)

            data = self._data[index : index + self.entry_size]

        r = self._unpack_entry(i, data)

        if self._lgt and i == 0:

            offset = revlogutils.offset_type(0, gettype(r[0]))

            r = (offset,) + r[1:]

        return r

    def _unpack_entry(self, rev, data):

        r = self.index_format.unpack(data)

        r = r + (

            0,

            0,

            revlog_constants.COMP_MODE_INLINE,

            revlog_constants.COMP_MODE_INLINE,

        )

        return r

    def pack_header(self, header):

        """pack header information as binary"""

        v_fmt = revlog_constants.INDEX_HEADER

        return v_fmt.pack(header)

    def entry_binary(self, rev):

        """return the raw binary string representing a revision"""

        entry = self[rev]

        p = revlog_constants.INDEX_ENTRY_V1.pack(*entry[:8])

        if rev == 0:

            p = p[revlog_constants.INDEX_HEADER.size :]

        return p

class IndexObject(BaseIndexObject):

    def __init__(self, data):

        assert len(data) % self.entry_size == 0, (

            len(data),

            self.entry_size,

            len(data) % self.entry_size,

        )

        self._data = data

        self._lgt = len(data) // self.entry_size

        self._extra = []

    def _calculate_index(self, i):

        return i * self.entry_size

    def __delitem__(self, i):

        if not isinstance(i, slice) or not i.stop == -1 or i.step is not None:

            raise ValueError(b"deleting slices only supports a:-1 with step 1")

        i = i.start

        self._check_index(i)

        self._stripnodes(i)

        if i < self._lgt:

            self._data = self._data[: i * self.entry_size]

            self._lgt = i

            self._extra = []

        else:

            self._extra = self._extra[: i - self._lgt]

class PersistentNodeMapIndexObject(IndexObject):

    """a Debug oriented class to test persistent nodemap

    We need a simple python object to test API and higher level behavior. See

    the Rust implementation for  more serious usage. This should be used only

    through the dedicated `devel.persistent-nodemap` config.

    """

    def nodemap_data_all(self):

        """Return bytes containing a full serialization of a nodemap

        The nodemap should be valid for the full set of revisions in the

        index."""

        return nodemaputil.persistent_data(self)

    def nodemap_data_incremental(self):

        """Return bytes containing a incremental update to persistent nodemap

        This containst the data for an append-only update of the data provided

        in the last call to `update_nodemap_data`.

        """

        if self._nm_root is None:

            return None

        docket = self._nm_docket

        changed, data = nodemaputil.update_persistent_data(

            self, self._nm_root, self._nm_max_idx, self._nm_docket.tip_rev

        )

        self._nm_root = self._nm_max_idx = self._nm_docket = None

        return docket, changed, data

    def update_nodemap_data(self, docket, nm_data):

        """provide full block of persisted binary data for a nodemap

        The data are expected to come from disk. See `nodemap_data_all` for a

        produceur of such data."""

        if nm_data is not None:

            self._nm_root, self._nm_max_idx = nodemaputil.parse_data(nm_data)

            if self._nm_root:

                self._nm_docket = docket

            else:

                self._nm_root = self._nm_max_idx = self._nm_docket = None

class InlinedIndexObject(BaseIndexObject):

    def __init__(self, data, inline=0):

        self._data = data

        self._lgt = self._inline_scan(None)

        self._inline_scan(self._lgt)

        self._extra = []

    def _inline_scan(self, lgt):

        off = 0

        if lgt is not None:

            self._offsets = [0] * lgt

        count = 0

        while off <= len(self._data) - self.entry_size:

            start = off + self.big_int_size

            (s,) = struct.unpack(

                b'>i',

                self._data[start : start + self.int_size],

            )

            if lgt is not None:

                self._offsets[count] = off

            count += 1

            off += self.entry_size + s

        if off != len(self._data):

            raise ValueError(b"corrupted data")

        return count

    def __delitem__(self, i):

        if not isinstance(i, slice) or not i.stop == -1 or i.step is not None:

            raise ValueError(b"deleting slices only supports a:-1 with step 1")

        i = i.start

        self._check_index(i)

        self._stripnodes(i)

        if i < self._lgt:

            self._offsets = self._offsets[:i]

            self._lgt = i

            self._extra = []

        else:

            self._extra = self._extra[: i - self._lgt]

    def _calculate_index(self, i):

        return self._offsets[i]

def parse_index2(data, inline, revlogv2=False):

    if not inline:

        cls = IndexObject2 if revlogv2 else IndexObject

        return cls(data), None

    cls = InlinedIndexObject

    return cls(data, inline), (0, data)

def parse_index_cl_v2(data):

    return IndexChangelogV2(data), None

class IndexObject2(IndexObject):

    index_format = revlog_constants.INDEX_ENTRY_V2

    def replace_sidedata_info(

        self,

        rev,

        sidedata_offset,

        sidedata_length,

        offset_flags,

        compression_mode,

    ):

        """

        Replace an existing index entry's sidedata offset and length with new

        ones.

        This cannot be used outside of the context of sidedata rewriting,

        inside the transaction that creates the revision `rev`.

        """

        if rev < 0:

            raise KeyError

        self._check_index(rev)

        if rev < self._lgt: