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

import functools

import os

import re

import stat

from typing import Generator

from .i18n import _

from .pycompat import getattr

from .thirdparty import attr

from .node import hex

from . import (

    changelog,

    error,

    manifest,

    policy,

    pycompat,

    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 _matchtrackedpath(path, 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

    path = decodedir(path)

    if path.startswith(b'data/'):

        return matcher(path[len(b'data/') : -len(b'.i')])

    elif path.startswith(b'meta/'):

        return matcher.visitdir(path[len(b'meta/') : -len(b'/00manifest.i')])

    raise error.ProgrammingError(b"cannot decode path %s" % path)

# 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_MAIN_EXT = (b'.i',)

REVLOG_FILES_OTHER_EXT = (