# store.py - repository store handling for Mercurial)

#

# Copyright 2008 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 collections

import functools

import os

import re

import stat

import typing

from typing import (

    Generator,

    List,

    Optional,

)

from .i18n import _

from .thirdparty import attr

# Force pytype to use the non-vendored package

if typing.TYPE_CHECKING:

    # noinspection PyPackageRequirements

    import attr

from .node import hex

from .revlogutils.constants import (

    INDEX_HEADER,

    KIND_CHANGELOG,

    KIND_FILELOG,

    KIND_MANIFESTLOG,

)

from . import (

    changelog,

    error,

    filelog,

    manifest,

    policy,

    pycompat,

    revlog as revlogmod,

    util,

    vfs as vfsmod,

)

from .utils import hashutil

parsers = policy.importmod('parsers')

# how much bytes should be read from fncache in one read

# It is done to prevent loading large fncache files into memory

fncache_chunksize = 10**6

def _match_tracked_entry(entry: "BaseStoreEntry", matcher):

    """parses a fncache entry and returns whether the entry is tracking a path

    matched by matcher or not.

    If matcher is None, returns True"""

    if matcher is None:

        return True

    # TODO: make this safe for other entry types.  Currently, the various

    #  store.data_entry generators only yield  RevlogStoreEntry, so the

    #  attributes do exist on `entry`.

    # pytype: disable=attribute-error

    if entry.is_filelog:

        return matcher(entry.target_id)

    elif entry.is_manifestlog:

        return matcher.visitdir(entry.target_id.rstrip(b'/'))

    # pytype: enable=attribute-error

    raise error.ProgrammingError(b"cannot process entry %r" % entry)

# This avoids a collision between a file named foo and a dir named

# foo.i or foo.d

def _encodedir(path):

    """

    >>> _encodedir(b'data/foo.i')

    'data/foo.i'

    >>> _encodedir(b'data/foo.i/bla.i')

    'data/foo.i.hg/bla.i'

    >>> _encodedir(b'data/foo.i.hg/bla.i')

    'data/foo.i.hg.hg/bla.i'

    >>> _encodedir(b'data/foo.i\\ndata/foo.i/bla.i\\ndata/foo.i.hg/bla.i\\n')

    'data/foo.i\\ndata/foo.i.hg/bla.i\\ndata/foo.i.hg.hg/bla.i\\n'

    """

    return (

        path.replace(b".hg/", b".hg.hg/")

        .replace(b".i/", b".i.hg/")

        .replace(b".d/", b".d.hg/")

    )

encodedir = getattr(parsers, 'encodedir', _encodedir)

def decodedir(path):

    """

    >>> decodedir(b'data/foo.i')

    'data/foo.i'

    >>> decodedir(b'data/foo.i.hg/bla.i')

    'data/foo.i/bla.i'

    >>> decodedir(b'data/foo.i.hg.hg/bla.i')

    'data/foo.i.hg/bla.i'

    """

    if b".hg/" not in path:

        return path

    return (

        path.replace(b".d.hg/", b".d/")

        .replace(b".i.hg/", b".i/")

        .replace(b".hg.hg/", b".hg/")

    )

def _reserved():

    """characters that are problematic for filesystems

    * ascii escapes (0..31)

    * ascii hi (126..255)

    * windows specials

    these characters will be escaped by encodefunctions

    """

    winreserved = [ord(x) for x in u'\\:*?"<>|']

    for x in range(32):

        yield x

    for x in range(126, 256):

        yield x

    for x in winreserved:

        yield x

def _buildencodefun():

    """

    >>> enc, dec = _buildencodefun()

    >>> enc(b'nothing/special.txt')

    'nothing/special.txt'

    >>> dec(b'nothing/special.txt')

    'nothing/special.txt'

    >>> enc(b'HELLO')

    '_h_e_l_l_o'

    >>> dec(b'_h_e_l_l_o')

    'HELLO'

    >>> enc(b'hello:world?')

    'hello~3aworld~3f'

    >>> dec(b'hello~3aworld~3f')

    'hello:world?'

    >>> enc(b'the\\x07quick\\xADshot')

    'the~07quick~adshot'

    >>> dec(b'the~07quick~adshot')

    'the\\x07quick\\xadshot'

    """

    e = b'_'

    xchr = pycompat.bytechr

    asciistr = list(map(xchr, range(127)))

    capitals = list(range(ord(b"A"), ord(b"Z") + 1))

    cmap = {x: x for x in asciistr}

    for x in _reserved():

        cmap[xchr(x)] = b"~%02x" % x

    for x in capitals + [ord(e)]:

        cmap[xchr(x)] = e + xchr(x).lower()

    dmap = {}

    for k, v in cmap.items():

        dmap[v] = k

    def decode(s):

        i = 0

        while i < len(s):

            for l in range(1, 4):

                try:

                    yield dmap[s[i : i + l]]

                    i += l

                    break

                except KeyError:

                    pass

            else:

                raise KeyError

    return (

        lambda s: b''.join([cmap[s[c : c + 1]] for c in range(len(s))]),

        lambda s: b''.join(list(decode(s))),

    )

_encodefname, _decodefname = _buildencodefun()

def encodefilename(s):

    """

    >>> encodefilename(b'foo.i/bar.d/bla.hg/hi:world?/HELLO')

    'foo.i.hg/bar.d.hg/bla.hg.hg/hi~3aworld~3f/_h_e_l_l_o'

    """

    return _encodefname(encodedir(s))

def decodefilename(s):

    """

    >>> decodefilename(b'foo.i.hg/bar.d.hg/bla.hg.hg/hi~3aworld~3f/_h_e_l_l_o')

    'foo.i/bar.d/bla.hg/hi:world?/HELLO'

    """

    return decodedir(_decodefname(s))

def _buildlowerencodefun():

    """

    >>> f = _buildlowerencodefun()

    >>> f(b'nothing/special.txt')

    'nothing/special.txt'

    >>> f(b'HELLO')

    'hello'

    >>> f(b'hello:world?')

    'hello~3aworld~3f'

    >>> f(b'the\\x07quick\\xADshot')

    'the~07quick~adshot'

    """

    xchr = pycompat.bytechr

    cmap = {xchr(x): xchr(x) for x in range(127)}

    for x in _reserved():

        cmap[xchr(x)] = b"~%02x" % x

    for x in range(ord(b"A"), ord(b"Z") + 1):

        cmap[xchr(x)] = xchr(x).lower()

    def lowerencode(s):

        return b"".join([cmap[c] for c in pycompat.iterbytestr(s)])

    return lowerencode

lowerencode = getattr(parsers, 'lowerencode', None) or _buildlowerencodefun()

# Windows reserved names: con, prn, aux, nul, com1..com9, lpt1..lpt9

_winres3 = (b'aux', b'con', b'prn', b'nul')  # length 3

_winres4 = (b'com', b'lpt')  # length 4 (with trailing 1..9)

def _auxencode(path, dotencode):

    """

    Encodes filenames containing names reserved by Windows or which end in

    period or space. Does not touch other single reserved characters c.

    Specifically, c in '\\:*?"<>|' or ord(c) <= 31 are *not* encoded here.

    Additionally encodes space or period at the beginning, if dotencode is

    True. Parameter path is assumed to be all lowercase.

    A segment only needs encoding if a reserved name appears as a

    basename (e.g. "aux", "aux.foo"). A directory or file named "foo.aux"

    doesn't need encoding.

    >>> s = b'.foo/aux.txt/txt.aux/con/prn/nul/foo.'

    >>> _auxencode(s.split(b'/'), True)

    ['~2efoo', 'au~78.txt', 'txt.aux', 'co~6e', 'pr~6e', 'nu~6c', 'foo~2e']

    >>> s = b'.com1com2/lpt9.lpt4.lpt1/conprn/com0/lpt0/foo.'

    >>> _auxencode(s.split(b'/'), False)

    ['.com1com2', 'lp~749.lpt4.lpt1', 'conprn', 'com0', 'lpt0', 'foo~2e']

    >>> _auxencode([b'foo. '], True)

    ['foo.~20']

    >>> _auxencode([b' .foo'], True)

    ['~20.foo']

    """

    for i, n in enumerate(path):

        if not n:

            continue

        if dotencode and n[0] in b'. ':

            n = b"~%02x" % ord(n[0:1]) + n[1:]

            path[i] = n

        else:

            l = n.find(b'.')

            if l == -1:

                l = len(n)

            if (l == 3 and n[:3] in _winres3) or (

                l == 4

                and n[3:4] <= b'9'

                and n[3:4] >= b'1'

                and n[:3] in _winres4

            ):

                # encode third letter ('aux' -> 'au~78')

                ec = b"~%02x" % ord(n[2:3])

                n = n[0:2] + ec + n[3:]

                path[i] = n

        if n[-1] in b'. ':

            # encode last period or space ('foo...' -> 'foo..~2e')

            path[i] = n[:-1] + b"~%02x" % ord(n[-1:])

    return path

_maxstorepathlen = 120

_dirprefixlen = 8

_maxshortdirslen = 8 * (_dirprefixlen + 1) - 4

def _hashencode(path, dotencode):

    digest = hex(hashutil.sha1(path).digest())

    le = lowerencode(path[5:]).split(b'/')  # skips prefix 'data/' or 'meta/'

    parts = _auxencode(le, dotencode)

    basename = parts[-1]

    _root, ext = os.path.splitext(basename)

    sdirs = []

    sdirslen = 0

    for p in parts[:-1]:

        d = p[:_dirprefixlen]

        if d[-1] in b'. ':

            # Windows can't access dirs ending in period or space

            d = d[:-1] + b'_'

        if sdirslen == 0:

            t = len(d)

        else:

            t = sdirslen + 1 + len(d)

            if t > _maxshortdirslen:

                break

        sdirs.append(d)

        sdirslen = t

    dirs = b'/'.join(sdirs)

    if len(dirs) > 0:

        dirs += b'/'

    res = b'dh/' + dirs + digest + ext

    spaceleft = _maxstorepathlen - len(res)

    if spaceleft > 0:

        filler = basename[:spaceleft]

        res = b'dh/' + dirs + filler + digest + ext

    return res

def _hybridencode(path, dotencode):

    """encodes path with a length limit

    Encodes all paths that begin with 'data/', according to the following.

    Default encoding (reversible):

    Encodes all uppercase letters 'X' as '_x'. All reserved or illegal

    characters are encoded as '~xx', where xx is the two digit hex code

    of the character (see encodefilename).

    Relevant path components consisting of Windows reserved filenames are

    masked by encoding the third character ('aux' -> 'au~78', see _auxencode).

    Hashed encoding (not reversible):

    If the default-encoded path is longer than _maxstorepathlen, a

    non-reversible hybrid hashing of the path is done instead.

    This encoding uses up to _dirprefixlen characters of all directory

    levels of the lowerencoded path, but not more levels than can fit into

    _maxshortdirslen.

    Then follows the filler followed by the sha digest of the full path.

    The filler is the beginning of the basename of the lowerencoded path

    (the basename is everything after the last path separator). The filler

    is as long as possible, filling in characters from the basename until

    the encoded path has _maxstorepathlen characters (or all chars of the

    basename have been taken).

    The extension (e.g. '.i' or '.d') is preserved.

    The string 'data/' at the beginning is replaced with 'dh/', if the hashed

    encoding was used.

    """

    path = encodedir(path)

    ef = _encodefname(path).split(b'/')

    res = b'/'.join(_auxencode(ef, dotencode))

    if len(res) > _maxstorepathlen:

        res = _hashencode(path, dotencode)

    return res

def _pathencode(path):

    de = encodedir(path)

    if len(path) > _maxstorepathlen:

        return _hashencode(de, True)

    ef = _encodefname(de).split(b'/')

    res = b'/'.join(_auxencode(ef, True))

    if len(res) > _maxstorepathlen:

        return _hashencode(de, True)

    return res

_pathencode = getattr(parsers, 'pathencode', _pathencode)

def _plainhybridencode(f):

    return _hybridencode(f, False)

def _calcmode(vfs):

    try:

        # files in .hg/ will be created using this mode

        mode = vfs.stat().st_mode

        # avoid some useless chmods

        if (0o777 & ~util.umask) == (0o777 & mode):

            mode = None

    except OSError:

        mode = None

    return mode

_data = [

    b'bookmarks',

    b'narrowspec',

    b'data',

    b'meta',

    b'00manifest.d',

    b'00manifest.i',

    b'00changelog.d',

    b'00changelog.i',

    b'phaseroots',

    b'obsstore',

    b'requires',

]

REVLOG_FILES_EXT = (

    b'.i',

    b'.idx',

    b'.d',

    b'.dat',

    b'.n',

    b'.nd',

    b'.sda',

)

# file extension that also use a `-SOMELONGIDHASH.ext` form

REVLOG_FILES_LONG_EXT = (

    b'.nd',

    b'.idx',

    b'.dat',

    b'.sda',

)

# files that are "volatile" and might change between listing and streaming

#

# note: the ".nd" file are nodemap data and won't "change" but they might be

# deleted.

REVLOG_FILES_VOLATILE_EXT = (b'.n', b'.nd')

# some exception to the above matching

#

# XXX This is currently not in use because of issue6542

EXCLUDED = re.compile(br'.*undo\.[^/]+\.(nd?|i)$')

def is_revlog(f, kind, st):

    if kind != stat.S_IFREG:

        return False

    if f.endswith(REVLOG_FILES_EXT):

        return True

    return False

def is_revlog_file(f):

    if f.endswith(REVLOG_FILES_EXT):

        return True

    return False

@attr.s(slots=True)

class StoreFile:

    """a file matching a store entry"""

    unencoded_path = attr.ib()

    _file_size = attr.ib(default=None)

    is_volatile = attr.ib(default=False)

    def file_size(self, vfs):

        if self._file_size is None:

            if vfs is None:

                msg = b"calling vfs-less file_size without prior call: %s"

                msg %= self.unencoded_path

                raise error.ProgrammingError(msg)

            try:

                self._file_size = vfs.stat(self.unencoded_path).st_size

            except FileNotFoundError:

                self._file_size = 0

        return self._file_size

    @property

    def has_size(self):

        return self._file_size is not None