# hg.py - repository classes for mercurial

#

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

# Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.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 os

import posixpath

import shutil

import stat

import typing

from .i18n import _

from .node import (

    hex,

    sha1nodeconstants,

    short,

)

from . import (

    bookmarks,

    bundlerepo,

    cmdutil,

    destutil,

    discovery,

    error,

    exchange,

    extensions,

    graphmod,

    httppeer,

    localrepo,

    lock,

    logcmdutil,

    logexchange,

    merge as mergemod,

    mergestate as mergestatemod,

    narrowspec,

    phases,

    requirements,

    scmutil,

    sshpeer,

    statichttprepo,

    ui as uimod,

    unionrepo,

    url,

    util,

    verify as verifymod,

    vfs as vfsmod,

)

from .interfaces import repository as repositorymod

from .utils import (

    hashutil,

    stringutil,

    urlutil,

)

if typing.TYPE_CHECKING:

    from typing import (

        List,

        Tuple,

    )

release = lock.release

# shared features

sharedbookmarks = b'bookmarks'

def addbranchrevs(lrepo, other, branches, revs, remotehidden=False):

    if hasattr(other, 'peer'):

        # a courtesy to callers using a localrepo for other

        peer = other.peer(remotehidden=remotehidden)

    else:

        peer = other

    hashbranch, branches = branches

    if not hashbranch and not branches:

        x = revs or None

        if revs:

            y = revs[0]

        else:

            y = None

        return x, y

    if revs:

        revs = list(revs)

    else:

        revs = []

    if not peer.capable(b'branchmap'):

        if branches:

            raise error.Abort(_(b"remote branch lookup not supported"))

        revs.append(hashbranch)

        return revs, revs[0]

    with peer.commandexecutor() as e:

        branchmap = e.callcommand(b'branchmap', {}).result()

    def primary(branch):

        if branch == b'.':

            if not lrepo:

                raise error.Abort(_(b"dirstate branch not accessible"))

            branch = lrepo.dirstate.branch()

        if branch in branchmap:

            revs.extend(hex(r) for r in reversed(branchmap[branch]))

            return True

        else:

            return False

    for branch in branches:

        if not primary(branch):

            raise error.RepoLookupError(_(b"unknown branch '%s'") % branch)

    if hashbranch:

        if not primary(hashbranch):

            revs.append(hashbranch)

    return revs, revs[0]

def _isfile(path):

    try:

        # we use os.stat() directly here instead of os.path.isfile()

        # because the latter started returning `False` on invalid path

        # exceptions starting in 3.8 and we care about handling

        # invalid paths specially here.

        st = os.stat(path)

    except ValueError as e:

        msg = stringutil.forcebytestr(e)

        raise error.Abort(_(b'invalid path %s: %s') % (path, msg))

    except OSError:

        return False

    else:

        return stat.S_ISREG(st.st_mode)

class LocalFactory:

    """thin wrapper to dispatch between localrepo and bundle repo"""

    @staticmethod

    def islocal(path: bytes) -> bool:

        path = util.expandpath(urlutil.urllocalpath(path))

        return not _isfile(path)

    @staticmethod

    def instance(ui, path, *args, **kwargs):

        path = util.expandpath(urlutil.urllocalpath(path))

        if _isfile(path):

            cls = bundlerepo

        else:

            cls = localrepo

        return cls.instance(ui, path, *args, **kwargs)

repo_schemes = {

    b'bundle': bundlerepo,

    b'union': unionrepo,

    b'file': LocalFactory,

}

peer_schemes = {

    b'http': httppeer,

    b'https': httppeer,

    b'ssh': sshpeer,

    b'static-http': statichttprepo,

}

def islocal(repo):

    '''return true if repo (or path pointing to repo) is local'''

    if isinstance(repo, bytes):

        u = urlutil.url(repo)

        scheme = u.scheme or b'file'

        if scheme in peer_schemes:

            cls = peer_schemes[scheme]

            cls.make_peer  # make sure we load the module

        elif scheme in repo_schemes:

            cls = repo_schemes[scheme]

            cls.instance  # make sure we load the module

        else:

            cls = LocalFactory

        if hasattr(cls, 'islocal'):

            return cls.islocal(repo)  # pytype: disable=module-attr

        return False

    repo.ui.deprecwarn(b"use obj.local() instead of islocal(obj)", b"6.4")

    return repo.local()

def openpath(ui, path, sendaccept=True):

    '''open path with open if local, url.open if remote'''

    pathurl = urlutil.url(path, parsequery=False, parsefragment=False)

    if pathurl.islocal():

        return util.posixfile(pathurl.localpath(), b'rb')

    else:

        return url.open(ui, path, sendaccept=sendaccept)

# a list of (ui, repo) functions called for wire peer initialization

wirepeersetupfuncs = []

def _setup_repo_or_peer(ui, obj, presetupfuncs=None):

    ui = getattr(obj, "ui", ui)

    for f in presetupfuncs or []:

        f(ui, obj)

    ui.log(b'extension', b'- executing reposetup hooks\n')

    with util.timedcm('all reposetup') as allreposetupstats:

        for name, module in extensions.extensions(ui):

            ui.log(b'extension', b'  - running reposetup for %s\n', name)

            hook = getattr(module, 'reposetup', None)

            if hook:

                with util.timedcm('reposetup %r', name) as stats:

                    hook(ui, obj)

                msg = b'  > reposetup for %s took %s\n'

                ui.log(b'extension', msg, name, stats)

    ui.log(b'extension', b'> all reposetup took %s\n', allreposetupstats)

    if not obj.local():

        for f in wirepeersetupfuncs:

            f(ui, obj)

def repository(

    ui,

    path=b'',

    create=False,

    presetupfuncs=None,

    intents=None,

    createopts=None,

):

    """return a repository object for the specified path"""

    scheme = urlutil.url(path).scheme

    if scheme is None:

        scheme = b'file'

    cls = repo_schemes.get(scheme)

    if cls is None:

        if scheme in peer_schemes:

            raise error.Abort(_(b"repository '%s' is not local") % path)

        cls = LocalFactory

    repo = cls.instance(

        ui,

        path,

        create,

        intents=intents,

        createopts=createopts,

    )

    _setup_repo_or_peer(ui, repo, presetupfuncs=presetupfuncs)

    return repo.filtered(b'visible')

def peer(

    uiorrepo,

    opts,

    path,

    create=False,

    intents=None,

    createopts=None,

    remotehidden=False,

):

    '''return a repository peer for the specified path'''

    ui = getattr(uiorrepo, 'ui', uiorrepo)

    rui = remoteui(uiorrepo, opts)

    if hasattr(path, 'url'):

        # this is already a urlutil.path object

        peer_path = path

    else:

        peer_path = urlutil.path(ui, None, rawloc=path, validate_path=False)

    scheme = peer_path.url.scheme  # pytype: disable=attribute-error

    if scheme in peer_schemes:

        cls = peer_schemes[scheme]

        peer = cls.make_peer(

            rui,

            peer_path,

            create,

            intents=intents,

            createopts=createopts,

            remotehidden=remotehidden,

        )

        _setup_repo_or_peer(rui, peer)

    else:

        # this is a repository

        repo_path = peer_path.loc  # pytype: disable=attribute-error

        if not repo_path:

            repo_path = peer_path.rawloc  # pytype: disable=attribute-error

        repo = repository(

            rui,

            repo_path,

            create,

            intents=intents,

            createopts=createopts,

        )

        peer = repo.peer(path=peer_path, remotehidden=remotehidden)

    return peer

def defaultdest(source):

    """return default destination of clone if none is given

    >>> defaultdest(b'foo')

    'foo'

    >>> defaultdest(b'/foo/bar')

    'bar'

    >>> defaultdest(b'/')

    ''

    >>> defaultdest(b'')

    ''

    >>> defaultdest(b'http://example.org/')

    ''

    >>> defaultdest(b'http://example.org/foo/')

    'foo'

    """

    path = urlutil.url(source).path

    if not path:

        return b''

    return os.path.basename(os.path.normpath(path))

def sharedreposource(repo):

    """Returns repository object for source repository of a shared repo.

    If repo is not a shared repository, returns None.

    """

    if repo.sharedpath == repo.path:

        return None

    if hasattr(repo, 'srcrepo') and repo.srcrepo:

        return repo.srcrepo

    # the sharedpath always ends in the .hg; we want the path to the repo

    source = repo.vfs.split(repo.sharedpath)[0]

    srcurl, branches = urlutil.parseurl(source)

    srcrepo = repository(repo.ui, srcurl)

    repo.srcrepo = srcrepo

    return srcrepo

def share(

    ui,

    source,

    dest=None,

    update=True,

    bookmarks=True,

    defaultpath=None,

    relative=False,

):

    '''create a shared repository'''

    not_local_msg = _(b'can only share local repositories')

    if hasattr(source, 'local'):

        if source.local() is None:

            raise error.Abort(not_local_msg)

    elif not islocal(source):

        # XXX why are we getting bytes here ?

        raise error.Abort(not_local_msg)

    if not dest:

        dest = defaultdest(source)

    else:

        dest = urlutil.get_clone_path_obj(ui, dest).loc

    if isinstance(source, bytes):

        source_path = urlutil.get_clone_path_obj(ui, source)

        srcrepo = repository(ui, source_path.loc)

        branches = (source_path.branch, [])

        rev, checkout = addbranchrevs(srcrepo, srcrepo, branches, None)

    else:

        srcrepo = source.local()

        checkout = None

    shareditems = set()

    if bookmarks:

        shareditems.add(sharedbookmarks)

    r = repository(

        ui,

        dest,

        create=True,

        createopts={

            b'sharedrepo': srcrepo,

            b'sharedrelative': relative,

            b'shareditems': shareditems,

        },

    )

    postshare(srcrepo, r, defaultpath=defaultpath)

    r = repository(ui, dest)

    _postshareupdate(r, update, checkout=checkout)

    return r

def _prependsourcehgrc(repo):

    """copies the source repo config and prepend it in current repo .hg/hgrc

    on unshare. This is only done if the share was perfomed using share safe

    method where we share config of source in shares"""

    srcvfs = vfsmod.vfs(repo.sharedpath)

    dstvfs = vfsmod.vfs(repo.path)

    if not srcvfs.exists(b'hgrc'):

        return

    currentconfig = b''

    if dstvfs.exists(b'hgrc'):

        currentconfig = dstvfs.read(b'hgrc')

    with dstvfs(b'hgrc', b'wb') as fp:

        sourceconfig = srcvfs.read(b'hgrc')

        fp.write(b"# Config copied from shared source\n")

        fp.write(sourceconfig)

        fp.write(b'\n')

        fp.write(currentconfig)

def unshare(ui, repo):

    """convert a shared repository to a normal one

    Copy the store data to the repo and remove the sharedpath data.

    Returns a new repository object representing the unshared repository.

    The passed repository object is not usable after this function is

    called.

    """

    with repo.lock():

        # we use locks here because if we race with commit, we

        # can end up with extra data in the cloned revlogs that's

        # not pointed to by changesets, thus causing verify to

        # fail

        destlock = copystore(ui, repo, repo.path)

        with destlock or util.nullcontextmanager():

            if requirements.SHARESAFE_REQUIREMENT in repo.requirements:

                # we were sharing .hg/hgrc of the share source with the current

                # repo. We need to copy that while unsharing otherwise it can

                # disable hooks and other checks

                _prependsourcehgrc(repo)

            sharefile = repo.vfs.join(b'sharedpath')

            util.rename(sharefile, sharefile + b'.old')

            repo.requirements.discard(requirements.SHARED_REQUIREMENT)

            repo.requirements.discard(requirements.RELATIVE_SHARED_REQUIREMENT)

            scmutil.writereporequirements(repo)

    # Removing share changes some fundamental properties of the repo instance.

    # So we instantiate a new repo object and operate on it rather than

    # try to keep the existing repo usable.

    newrepo = repository(repo.baseui, repo.root, create=False)

    # TODO: figure out how to access subrepos that exist, but were previously

    #       removed from .hgsub

    c = newrepo[b'.']

    subs = c.substate

    for s in sorted(subs):

        c.sub(s).unshare()

    localrepo.poisonrepository(repo)

    return newrepo

def postshare(sourcerepo, destrepo, defaultpath=None):

    """Called after a new shared repo is created.

    The new repo only has a requirements file and pointer to the source.

    This function configures additional shared data.

    Extensions can wrap this function and write additional entries to

    destrepo/.hg/shared to indicate additional pieces of data to be shared.

    """

    default = defaultpath or sourcerepo.ui.config(b'paths', b'default')

    if default:

        template = b'[paths]\ndefault = %s\n'

        destrepo.vfs.write(b'hgrc', util.tonativeeol(template % default))

    if requirements.NARROW_REQUIREMENT in sourcerepo.requirements:

        with destrepo.wlock(), destrepo.lock(), destrepo.transaction(

            b"narrow-share"

        ):

            narrowspec.copytoworkingcopy(destrepo)

def _postshareupdate(repo, update, checkout=None):