# localrepo.py - read/write repository class for mercurial

# coding: utf-8

#

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

import functools

import os

import random

import re

import sys

import time

import weakref

from concurrent import futures

from typing import (

    Optional,

)

from .i18n import _

from .node import (

    bin,

    hex,

    nullrev,

    sha1nodeconstants,

    short,

)

from . import (

    bookmarks,

    branchmap,

    bundle2,

    bundlecaches,

    changegroup,

    color,

    commit,

    context,

    dirstate,

    discovery,

    encoding,

    error,

    exchange,

    extensions,

    filelog,

    hook,

    lock as lockmod,

    match as matchmod,

    mergestate as mergestatemod,

    mergeutil,

    namespaces,

    narrowspec,

    obsolete,

    pathutil,

    phases,

    policy,

    pushkey,

    pycompat,

    rcutil,

    repoview,

    requirements as requirementsmod,

    revlog,

    revset,

    revsetlang,

    scmutil,

    sparse,

    store as storemod,

    subrepoutil,

    tags as tagsmod,

    transaction,

    txnutil,

    util,

    vfs as vfsmod,

    wireprototypes,

)

from .interfaces import (

    repository,

    util as interfaceutil,

)

from .utils import (

    hashutil,

    procutil,

    stringutil,

    urlutil,

)

from .revlogutils import (

    concurrency_checker as revlogchecker,

    constants as revlogconst,

    sidedata as sidedatamod,

)

release = lockmod.release

urlerr = util.urlerr

urlreq = util.urlreq

RE_SKIP_DIRSTATE_ROLLBACK = re.compile(

    b"^((dirstate|narrowspec.dirstate).*|branch$)"

)

# set of (path, vfs-location) tuples. vfs-location is:

# - 'plain for vfs relative paths

# - '' for svfs relative paths

_cachedfiles = set()

class _basefilecache(scmutil.filecache):

    """All filecache usage on repo are done for logic that should be unfiltered"""

    def __get__(self, repo, type=None):

        if repo is None:

            return self

        # proxy to unfiltered __dict__ since filtered repo has no entry

        unfi = repo.unfiltered()

        try:

            return unfi.__dict__[self.sname]

        except KeyError:

            pass

        return super(_basefilecache, self).__get__(unfi, type)

    def set(self, repo, value):

        return super(_basefilecache, self).set(repo.unfiltered(), value)

class repofilecache(_basefilecache):

    """filecache for files in .hg but outside of .hg/store"""

    def __init__(self, *paths):

        super(repofilecache, self).__init__(*paths)

        for path in paths:

            _cachedfiles.add((path, b'plain'))

    def join(self, obj, fname):

        return obj.vfs.join(fname)

class storecache(_basefilecache):

    """filecache for files in the store"""

    def __init__(self, *paths):

        super(storecache, self).__init__(*paths)

        for path in paths:

            _cachedfiles.add((path, b''))

    def join(self, obj, fname):

        return obj.sjoin(fname)

class changelogcache(storecache):

    """filecache for the changelog"""

    def __init__(self):

        super(changelogcache, self).__init__()

        _cachedfiles.add((b'00changelog.i', b''))

        _cachedfiles.add((b'00changelog.n', b''))

    def tracked_paths(self, obj):

        paths = [self.join(obj, b'00changelog.i')]

        if obj.store.opener.options.get(b'persistent-nodemap', False):

            paths.append(self.join(obj, b'00changelog.n'))

        return paths

class manifestlogcache(storecache):

    """filecache for the manifestlog"""

    def __init__(self):

        super(manifestlogcache, self).__init__()

        _cachedfiles.add((b'00manifest.i', b''))

        _cachedfiles.add((b'00manifest.n', b''))

    def tracked_paths(self, obj):

        paths = [self.join(obj, b'00manifest.i')]

        if obj.store.opener.options.get(b'persistent-nodemap', False):

            paths.append(self.join(obj, b'00manifest.n'))

        return paths

class mixedrepostorecache(_basefilecache):

    """filecache for a mix files in .hg/store and outside"""

    def __init__(self, *pathsandlocations):

        # scmutil.filecache only uses the path for passing back into our

        # join(), so we can safely pass a list of paths and locations

        super(mixedrepostorecache, self).__init__(*pathsandlocations)

        _cachedfiles.update(pathsandlocations)

    def join(self, obj, fnameandlocation):

        fname, location = fnameandlocation

        if location == b'plain':

            return obj.vfs.join(fname)

        else:

            if location != b'':

                raise error.ProgrammingError(

                    b'unexpected location: %s' % location

                )

            return obj.sjoin(fname)

def isfilecached(repo, name):

    """check if a repo has already cached "name" filecache-ed property

    This returns (cachedobj-or-None, iscached) tuple.

    """

    cacheentry = repo.unfiltered()._filecache.get(name, None)

    if not cacheentry:

        return None, False

    return cacheentry.obj, True

class unfilteredpropertycache(util.propertycache):

    """propertycache that apply to unfiltered repo only"""

    def __get__(self, repo, type=None):

        unfi = repo.unfiltered()

        if unfi is repo:

            return super(unfilteredpropertycache, self).__get__(unfi)

        return getattr(unfi, self.name)

class filteredpropertycache(util.propertycache):

    """propertycache that must take filtering in account"""

    def cachevalue(self, obj, value):

        object.__setattr__(obj, self.name, value)

def hasunfilteredcache(repo, name):

    """check if a repo has an unfilteredpropertycache value for <name>"""

    return name in vars(repo.unfiltered())

def unfilteredmethod(orig):

    """decorate method that always need to be run on unfiltered version"""

    @functools.wraps(orig)

    def wrapper(repo, *args, **kwargs):

        return orig(repo.unfiltered(), *args, **kwargs)

    return wrapper

moderncaps = {

    b'lookup',

    b'branchmap',

    b'pushkey',

    b'known',

    b'getbundle',

    b'unbundle',

}

legacycaps = moderncaps.union({b'changegroupsubset'})

@interfaceutil.implementer(repository.ipeercommandexecutor)

class localcommandexecutor:

    def __init__(self, peer):

        self._peer = peer

        self._sent = False

        self._closed = False

    def __enter__(self):

        return self

    def __exit__(self, exctype, excvalue, exctb):

        self.close()

    def callcommand(self, command, args):

        if self._sent:

            raise error.ProgrammingError(

                b'callcommand() cannot be used after sendcommands()'

            )

        if self._closed:

            raise error.ProgrammingError(

                b'callcommand() cannot be used after close()'

            )

        # We don't need to support anything fancy. Just call the named

        # method on the peer and return a resolved future.

        fn = getattr(self._peer, pycompat.sysstr(command))

        f = futures.Future()

        try:

            result = fn(**pycompat.strkwargs(args))

        except Exception:

            pycompat.future_set_exception_info(f, sys.exc_info()[1:])

        else:

            f.set_result(result)

        return f

    def sendcommands(self):

        self._sent = True

    def close(self):

        self._closed = True

@interfaceutil.implementer(repository.ipeercommands)

class localpeer(repository.peer):

    '''peer for a local repo; reflects only the most recent API'''

    def __init__(self, repo, caps=None, path=None, remotehidden=False):

        super(localpeer, self).__init__(

            repo.ui, path=path, remotehidden=remotehidden

        )

        if caps is None:

            caps = moderncaps.copy()

        if remotehidden:

            self._repo = repo.filtered(b'served.hidden')

        else:

            self._repo = repo.filtered(b'served')

        if repo._wanted_sidedata:

            formatted = bundle2.format_remote_wanted_sidedata(repo)

            caps.add(b'exp-wanted-sidedata=' + formatted)

        self._caps = repo._restrictcapabilities(caps)

    # Begin of _basepeer interface.

    def url(self):

        return self._repo.url()

    def local(self):

        return self._repo

    def canpush(self):

        return True

    def close(self):

        self._repo.close()

    # End of _basepeer interface.

    # Begin of _basewirecommands interface.

    def branchmap(self):

        return self._repo.branchmap()

    def capabilities(self):

        return self._caps

    def get_cached_bundle_inline(self, path):

        # not needed with local peer

        raise NotImplementedError

    def clonebundles(self):

        return bundlecaches.get_manifest(self._repo)

    def debugwireargs(self, one, two, three=None, four=None, five=None):

        """Used to test argument passing over the wire"""

        return b"%s %s %s %s %s" % (

            one,

            two,

            pycompat.bytestr(three),

            pycompat.bytestr(four),

            pycompat.bytestr(five),

        )

    def getbundle(

        self,

        source,

        heads=None,

        common=None,

        bundlecaps=None,

        remote_sidedata=None,

        **kwargs,

    ):

        chunks = exchange.getbundlechunks(

            self._repo,

            source,

            heads=heads,

            common=common,

            bundlecaps=bundlecaps,

            remote_sidedata=remote_sidedata,

            **kwargs,

        )[1]

        cb = util.chunkbuffer(chunks)

        if exchange.bundle2requested(bundlecaps):

            # When requesting a bundle2, getbundle returns a stream to make the

            # wire level function happier. We need to build a proper object

            # from it in local peer.

            return bundle2.getunbundler(self.ui, cb)

        else:

            return changegroup.getunbundler(b'01', cb, None)

    def heads(self):