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

#

# Copyright 2005-2007 Matt Mackall <mpm@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 absolute_import

import errno

import os

import random

import sys

import time

import weakref

from .i18n import _

from .node import (

    bin,

    hex,

    nullid,

    nullrev,

    short,

)

from .pycompat import (

    delattr,

    getattr,

)

from . import (

    bookmarks,

    branchmap,

    bundle2,

    changegroup,

    color,

    context,

    dirstate,

    dirstateguard,

    discovery,

    encoding,

    error,

    exchange,

    extensions,

    filelog,

    hook,

    lock as lockmod,

    match as matchmod,

    merge as mergemod,

    mergeutil,

    namespaces,

    narrowspec,

    obsolete,

    pathutil,

    phases,

    pushkey,

    pycompat,

    repoview,

    revset,

    revsetlang,

    scmutil,

    sparse,

    store as storemod,

    subrepoutil,

    tags as tagsmod,

    transaction,

    txnutil,

    util,

    vfs as vfsmod,

)

from .interfaces import (

    repository,

    util as interfaceutil,

)

from .utils import (

    hashutil,

    procutil,

    stringutil,

)

from .revlogutils import constants as revlogconst

release = lockmod.release

urlerr = util.urlerr

urlreq = util.urlreq

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

    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(object):

    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 = pycompat.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):

        super(localpeer, self).__init__()

        if caps is None:

            caps = moderncaps.copy()

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

        self.ui = repo.ui

        self._caps = repo._restrictcapabilities(caps)

    # Begin of _basepeer interface.

    def url(self):

        return self._repo.url()

    def local(self):

        return self._repo

    def peer(self):

        return self

    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 clonebundles(self):

        return self._repo.tryread(b'clonebundles.manifest')

    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, **kwargs

    ):

        chunks = exchange.getbundlechunks(

            self._repo,

            source,

            heads=heads,

            common=common,

            bundlecaps=bundlecaps,

            **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):

        return self._repo.heads()

    def known(self, nodes):

        return self._repo.known(nodes)

    def listkeys(self, namespace):

        return self._repo.listkeys(namespace)

    def lookup(self, key):

        return self._repo.lookup(key)

    def pushkey(self, namespace, key, old, new):

        return self._repo.pushkey(namespace, key, old, new)

    def stream_out(self):

        raise error.Abort(_(b'cannot perform stream clone against local peer'))

    def unbundle(self, bundle, heads, url):

        """apply a bundle on a repo

        This function handles the repo locking itself."""

        try:

            try:

                bundle = exchange.readbundle(self.ui, bundle, None)

                ret = exchange.unbundle(self._repo, bundle, heads, b'push', url)

                if util.safehasattr(ret, b'getchunks'):

                    # This is a bundle20 object, turn it into an unbundler.

                    # This little dance should be dropped eventually when the

                    # API is finally improved.

                    stream = util.chunkbuffer(ret.getchunks())

                    ret = bundle2.getunbundler(self.ui, stream)

                return ret

            except Exception as exc:

                # If the exception contains output salvaged from a bundle2

                # reply, we need to make sure it is printed before continuing

                # to fail. So we build a bundle2 with such output and consume

                # it directly.

                #

                # This is not very elegant but allows a "simple" solution for

                # issue4594

                output = getattr(exc, '_bundle2salvagedoutput', ())

                if output:

                    bundler = bundle2.bundle20(self._repo.ui)

                    for out in output:

                        bundler.addpart(out)

                    stream = util.chunkbuffer(bundler.getchunks())

                    b = bundle2.getunbundler(self.ui, stream)

                    bundle2.processbundle(self._repo, b)

                raise

        except error.PushRaced as exc:

            raise error.ResponseError(

                _(b'push failed:'), stringutil.forcebytestr(exc)

            )

    # End of _basewirecommands interface.

    # Begin of peer interface.

    def commandexecutor(self):

        return localcommandexecutor(self)

    # End of peer interface.