# manifest.py - manifest revision class for mercurial

#

# Copyright 2005-2007 Olivia Mackall <olivia@selenic.com>

#

# 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 annotations

import heapq

import itertools

import struct

import typing

import weakref

from typing import (

    ByteString,

    Callable,

    Collection,

    Dict,

    Iterable,

    Iterator,

    List,

    Optional,

    Set,

    Tuple,

    Union,

    cast,

)

from .i18n import _

from .node import (

    bin,

    hex,

    nullrev,

)

from . import (

    encoding,

    error,

    match as matchmod,

    mdiff,

    pathutil,

    policy,

    pycompat,

    revlog,

    util,

)

from .interfaces import (

    repository,

    util as interfaceutil,

)

from .revlogutils import (

    constants as revlog_constants,

)

parsers = policy.importmod('parsers')

propertycache = util.propertycache

# Allow tests to more easily test the alternate path in manifestdict.fastdelta()

FASTDELTA_TEXTDIFF_THRESHOLD = 1000

def _parse(nodelen, data: bytes):

    # This method does a little bit of excessive-looking

    # precondition checking. This is so that the behavior of this

    # class exactly matches its C counterpart to try and help

    # prevent surprise breakage for anyone that develops against

    # the pure version.

    if data and data[-1:] != b'\n':

        raise ValueError(b'Manifest did not end in a newline.')

    prev = None

    for l in data.splitlines():

        if prev is not None and prev > l:

            raise ValueError(b'Manifest lines not in sorted order.')

        prev = l

        f, n = l.split(b'\0')

        nl = len(n)

        flags = n[-1:]

        if flags in _manifestflags:

            n = n[:-1]

            nl -= 1

        else:

            flags = b''

        if nl != 2 * nodelen:

            raise ValueError(b'Invalid manifest line')

        yield f, bin(n), flags

def _text(it):

    files = []

    lines = []

    for f, n, fl in it:

        files.append(f)

        # if this is changed to support newlines in filenames,

        # be sure to check the templates/ dir again (especially *-raw.tmpl)

        lines.append(b"%s\0%s%s\n" % (f, hex(n), fl))

    _checkforbidden(files)

    return b''.join(lines)

class lazymanifestiter:

    def __init__(self, lm: '_LazyManifest') -> None:

        self.pos = 0

        self.lm = lm

    def __iter__(self) -> 'lazymanifestiter':

        return self

    def next(self) -> bytes:

        try:

            data, pos = self.lm._get(self.pos)

        except IndexError:

            raise StopIteration

        if pos == -1:

            assert isinstance(data, tuple)

            self.pos += 1

            return data[0]

        assert isinstance(data, bytes)

        self.pos += 1

        zeropos = data.find(b'\x00', pos)

        return data[pos:zeropos]

    __next__ = next

class lazymanifestiterentries:

    def __init__(self, lm: '_LazyManifest') -> None:

        self.lm = lm

        self.pos = 0

    def __iter__(self) -> 'lazymanifestiterentries':

        return self

    def next(self) -> Tuple[bytes, bytes, bytes]:

        try:

            data, pos = self.lm._get(self.pos)

        except IndexError:

            raise StopIteration

        if pos == -1:

            assert isinstance(data, tuple)

            self.pos += 1

            return data

        assert isinstance(data, bytes)

        zeropos = data.find(b'\x00', pos)

        nlpos = data.find(b'\n', pos)

        if zeropos == -1 or nlpos == -1 or nlpos < zeropos:

            raise error.StorageError(b'Invalid manifest line')

        flags = data[nlpos - 1 : nlpos]

        if flags in _manifestflags:

            hlen = nlpos - zeropos - 2

        else:

            hlen = nlpos - zeropos - 1

            flags = b''

        if hlen != 2 * self.lm._nodelen:

            raise error.StorageError(b'Invalid manifest line')

        hashval = unhexlify(

            data, self.lm.extrainfo[self.pos], zeropos + 1, hlen

        )

        self.pos += 1

        return (data[pos:zeropos], hashval, flags)

    __next__ = next

def unhexlify(data: bytes, extra: int, pos, length: int):

    s = bin(data[pos : pos + length])

    if extra:

        s += bytes([extra & 0xFF])

    return s

def _cmp(a, b):

    return (a > b) - (a < b)

_manifestflags = {b'', b'l', b't', b'x'}

class _LazyManifest:

    """A pure python manifest backed by a byte string.  It is supplimented with

    internal lists as it is modified, until it is compacted back to a pure byte

    string.

    ``data`` is the initial manifest data.

    ``positions`` is a list of offsets, one per manifest entry.  Positive

    values are offsets into ``data``, negative values are offsets into the

    ``extradata`` list.  When an entry is removed, its entry is dropped from

    ``positions``.  The values are encoded such that when walking the list and

    indexing into ``data`` or ``extradata`` as appropriate, the entries are

    sorted by filename.

    ``extradata`` is a list of (key, hash, flags) for entries that were added or

    modified since the manifest was created or compacted.

    """

    def __init__(

        self,

        nodelen: int,

        data: bytes,

        positions=None,

        extrainfo=None,

        extradata=None,

        hasremovals: bool = False,

    ):

        self._nodelen = nodelen

        if positions is None:

            self.positions = self.findlines(data)

            self.extrainfo = [0] * len(self.positions)

            self.data = data

            self.extradata = []

            self.hasremovals = False

        else:

            self.positions = positions[:]

            self.extrainfo = extrainfo[:]

            self.extradata = extradata[:]

            self.data = data

            self.hasremovals = hasremovals

    def findlines(self, data: bytes) -> List[int]:

        if not data:

            return []

        pos = data.find(b"\n")

        if pos == -1 or data[-1:] != b'\n':

            raise ValueError(b"Manifest did not end in a newline.")

        positions = [0]

        prev = data[: data.find(b'\x00')]

        while pos < len(data) - 1 and pos != -1:

            positions.append(pos + 1)

            nexts = data[pos + 1 : data.find(b'\x00', pos + 1)]

            if nexts < prev:

                raise ValueError(b"Manifest lines not in sorted order.")

            prev = nexts

            pos = data.find(b"\n", pos + 1)

        return positions

    def _get(

        self, index: int

    ) -> Tuple[Union[bytes, Tuple[bytes, bytes, bytes]], int]:

        # get the position encoded in pos:

        #   positive number is an index in 'data'

        #   negative number is in extrapieces

        pos = self.positions[index]

        if pos >= 0:

            return self.data, pos

        return self.extradata[-pos - 1], -1

    def _getkey(self, pos) -> bytes:

        if pos >= 0:

            return self.data[pos : self.data.find(b'\x00', pos + 1)]

        return self.extradata[-pos - 1][0]

    def bsearch(self, key: bytes) -> int:

        first = 0

        last = len(self.positions) - 1

        while first <= last:

            midpoint = (first + last) // 2

            nextpos = self.positions[midpoint]

            candidate = self._getkey(nextpos)

            r = _cmp(key, candidate)

            if r == 0:

                return midpoint

            else:

                if r < 0:

                    last = midpoint - 1

                else:

                    first = midpoint + 1

        return -1

    def bsearch2(self, key: bytes) -> Tuple[int, bool]:

        # same as the above, but will always return the position

        # done for performance reasons

        first = 0

        last = len(self.positions) - 1

        while first <= last:

            midpoint = (first + last) // 2

            nextpos = self.positions[midpoint]

            candidate = self._getkey(nextpos)

            r = _cmp(key, candidate)

            if r == 0:

                return (midpoint, True)

            else:

                if r < 0:

                    last = midpoint - 1

                else:

                    first = midpoint + 1

        return (first, False)

    def __contains__(self, key: bytes) -> bool:

        return self.bsearch(key) != -1

    def __getitem__(self, key: bytes) -> Tuple[bytes, bytes]:

        if not isinstance(key, bytes):

            raise TypeError(b"getitem: manifest keys must be a bytes.")

        needle = self.bsearch(key)

        if needle == -1:

            raise KeyError

        data, pos = self._get(needle)

        if pos == -1:

            assert isinstance(data, tuple)

            return (data[1], data[2])

        assert isinstance(data, bytes)

        zeropos = data.find(b'\x00', pos)

        nlpos = data.find(b'\n', zeropos)

        assert 0 <= needle <= len(self.positions)

        assert len(self.extrainfo) == len(self.positions)

        if zeropos == -1 or nlpos == -1 or nlpos < zeropos:

            raise error.StorageError(b'Invalid manifest line')

        hlen = nlpos - zeropos - 1

        flags = data[nlpos - 1 : nlpos]

        if flags in _manifestflags:

            hlen -= 1

        else:

            flags = b''

        if hlen != 2 * self._nodelen:

            raise error.StorageError(b'Invalid manifest line')

        hashval = unhexlify(data, self.extrainfo[needle], zeropos + 1, hlen)

        return (hashval, flags)

    def __delitem__(self, key: bytes) -> None:

        needle, found = self.bsearch2(key)

        if not found:

            raise KeyError

        cur = self.positions[needle]

        self.positions = self.positions[:needle] + self.positions[needle + 1 :]

        self.extrainfo = self.extrainfo[:needle] + self.extrainfo[needle + 1 :]

        if cur >= 0:

            # This does NOT unsort the list as far as the search functions are

            # concerned, as they only examine lines mapped by self.positions.

            self.data = self.data[:cur] + b'\x00' + self.data[cur + 1 :]

            self.hasremovals = True

    def __setitem__(self, key: bytes, value: Tuple[bytes, bytes]):

        if not isinstance(key, bytes):

            raise TypeError(b"setitem: manifest keys must be a byte string.")

        if not isinstance(value, tuple) or len(value) != 2:

            raise TypeError(

                b"Manifest values must be a tuple of (node, flags)."

            )

        hashval = value[0]

        if not isinstance(hashval, bytes) or len(hashval) not in (20, 32):

            raise TypeError(b"node must be a 20-byte or 32-byte byte string")

        flags = value[1]

        if not isinstance(flags, bytes) or len(flags) > 1:

            raise TypeError(b"flags must a 0 or 1 byte string, got %r", flags)

        needle, found = self.bsearch2(key)

        if found:

            # put the item

            pos = self.positions[needle]

            if pos < 0:

                self.extradata[-pos - 1] = (key, hashval, value[1])

            else:

                # just don't bother

                self.extradata.append((key, hashval, value[1]))

                self.positions[needle] = -len(self.extradata)

        else:

            # not found, put it in with extra positions

            self.extradata.append((key, hashval, value[1]))

            self.positions = (

                self.positions[:needle]

                + [-len(self.extradata)]

                + self.positions[needle:]

            )

            self.extrainfo = (

                self.extrainfo[:needle] + [0] + self.extrainfo[needle:]

            )

    def copy(self) -> '_LazyManifest':

        # XXX call _compact like in C?

        return _lazymanifest(

            self._nodelen,

            self.data,

            self.positions,

            self.extrainfo,

            self.extradata,

            self.hasremovals,

        )

    def _compact(self) -> None:

        # hopefully not called TOO often

        if len(self.extradata) == 0 and not self.hasremovals:

            return

        l = []

        i = 0

        offset = 0

        self.extrainfo = [0] * len(self.positions)

        while i < len(self.positions):

            if self.positions[i] >= 0:

                cur = self.positions[i]

                last_cut = cur

                # Collect all contiguous entries in the buffer at the current

                # offset, breaking out only for added/modified items held in

                # extradata, or a deleted line prior to the next position.

                while True:

                    self.positions[i] = offset

                    i += 1

                    if i == len(self.positions) or self.positions[i] < 0:

                        break

                    # A removed file has no positions[] entry, but does have an

                    # overwritten first byte.  Break out and find the end of the

                    # current good entry/entries if there is a removed file

                    # before the next position.

                    if (

                        self.hasremovals

                        and self.data.find(b'\n\x00', cur, self.positions[i])

                        != -1

                    ):

                        break

                    offset += self.positions[i] - cur

                    cur = self.positions[i]

                end_cut = self.data.find(b'\n', cur)

                if end_cut != -1:

                    end_cut += 1

                offset += end_cut - cur

                l.append(self.data[last_cut:end_cut])

            else:

                while i < len(self.positions) and self.positions[i] < 0:

                    cur = self.positions[i]

                    t = self.extradata[-cur - 1]

                    l.append(self._pack(t))

                    self.positions[i] = offset

                    # Hashes are either 20 bytes (old sha1s) or 32

                    # bytes (new non-sha1).

                    hlen = 20

                    if len(t[1]) > 25:

                        hlen = 32

                    if len(t[1]) > hlen:

                        self.extrainfo[i] = ord(t[1][hlen + 1])

                    offset += len(l[-1])

                    i += 1

        self.data = b''.join(l)

        self.hasremovals = False

        self.extradata = []

    def _pack(self, d: Tuple[bytes, bytes, bytes]) -> bytes:

        n = d[1]

        assert len(n) in (20, 32)

        return d[0] + b'\x00' + hex(n) + d[2] + b'\n'

    def text(self) -> ByteString:

        self._compact()

        return self.data

    def diff(

        self, m2: '_LazyManifest', clean: bool = False

    ) -> Dict[

        bytes,

        Optional[

            Tuple[Tuple[Optional[bytes], bytes], Tuple[Optional[bytes], bytes]]

        ],

    ]:

        '''Finds changes between the current manifest and m2.'''

        # XXX think whether efficiency matters here

        diff = {}

        for fn, e1, flags in self.iterentries():

            if fn not in m2:

                diff[fn] = (e1, flags), (None, b'')

            else:

                e2 = m2[fn]

                if (e1, flags) != e2:

                    diff[fn] = (e1, flags), e2

                elif clean:

                    diff[fn] = None

        for fn, e2, flags in m2.iterentries():

            if fn not in self:

                diff[fn] = (None, b''), (e2, flags)

        return diff

    def iterentries(self) -> lazymanifestiterentries: