# exchange.py - utility to exchange data between repos.

#

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

from .i18n import _

from .node import (

    hex,

    nullid,

)

from . import (

    bookmarks as bookmod,

    bundle2,

    changegroup,

    discovery,

    error,

    lock as lockmod,

    obsolete,

    phases,

    pushkey,

    pycompat,

    scmutil,

    sslutil,

    streamclone,

    url as urlmod,

    util,

)

urlerr = util.urlerr

urlreq = util.urlreq

# Maps bundle version human names to changegroup versions.

_bundlespeccgversions = {'v1': '01',

                         'v2': '02',

                         'packed1': 's1',

                         'bundle2': '02', #legacy

                        }

# Compression engines allowed in version 1. THIS SHOULD NEVER CHANGE.

_bundlespecv1compengines = {'gzip', 'bzip2', 'none'}

def parsebundlespec(repo, spec, strict=True, externalnames=False):

    """Parse a bundle string specification into parts.

    Bundle specifications denote a well-defined bundle/exchange format.

    The content of a given specification should not change over time in

    order to ensure that bundles produced by a newer version of Mercurial are

    readable from an older version.

    The string currently has the form:

       <compression>-<type>[;<parameter0>[;<parameter1>]]

    Where <compression> is one of the supported compression formats

    and <type> is (currently) a version string. A ";" can follow the type and

    all text afterwards is interpreted as URI encoded, ";" delimited key=value

    pairs.

    If ``strict`` is True (the default) <compression> is required. Otherwise,

    it is optional.

    If ``externalnames`` is False (the default), the human-centric names will

    be converted to their internal representation.

    Returns a 3-tuple of (compression, version, parameters). Compression will

    be ``None`` if not in strict mode and a compression isn't defined.

    An ``InvalidBundleSpecification`` is raised when the specification is

    not syntactically well formed.

    An ``UnsupportedBundleSpecification`` is raised when the compression or

    bundle type/version is not recognized.

    Note: this function will likely eventually return a more complex data

    structure, including bundle2 part information.

    """

    def parseparams(s):

        if ';' not in s:

            return s, {}

        params = {}

        version, paramstr = s.split(';', 1)

        for p in paramstr.split(';'):

            if '=' not in p:

                raise error.InvalidBundleSpecification(

                    _('invalid bundle specification: '

                      'missing "=" in parameter: %s') % p)

            key, value = p.split('=', 1)

            key = urlreq.unquote(key)

            value = urlreq.unquote(value)

            params[key] = value

        return version, params

    if strict and '-' not in spec:

        raise error.InvalidBundleSpecification(

                _('invalid bundle specification; '

                  'must be prefixed with compression: %s') % spec)

    if '-' in spec:

        compression, version = spec.split('-', 1)

        if compression not in util.compengines.supportedbundlenames:

            raise error.UnsupportedBundleSpecification(

                    _('%s compression is not supported') % compression)

        version, params = parseparams(version)

        if version not in _bundlespeccgversions:

            raise error.UnsupportedBundleSpecification(

                    _('%s is not a recognized bundle version') % version)

    else:

        # Value could be just the compression or just the version, in which

        # case some defaults are assumed (but only when not in strict mode).

        assert not strict

        spec, params = parseparams(spec)

        if spec in util.compengines.supportedbundlenames:

            compression = spec

            version = 'v1'

            # Generaldelta repos require v2.

            if 'generaldelta' in repo.requirements:

                version = 'v2'

            # Modern compression engines require v2.

            if compression not in _bundlespecv1compengines:

                version = 'v2'

        elif spec in _bundlespeccgversions:

            if spec == 'packed1':

                compression = 'none'

            else:

                compression = 'bzip2'

            version = spec

        else:

            raise error.UnsupportedBundleSpecification(

                    _('%s is not a recognized bundle specification') % spec)

    # Bundle version 1 only supports a known set of compression engines.

    if version == 'v1' and compression not in _bundlespecv1compengines:

        raise error.UnsupportedBundleSpecification(

            _('compression engine %s is not supported on v1 bundles') %

            compression)

    # The specification for packed1 can optionally declare the data formats

    # required to apply it. If we see this metadata, compare against what the

    # repo supports and error if the bundle isn't compatible.

    if version == 'packed1' and 'requirements' in params:

        requirements = set(params['requirements'].split(','))

        missingreqs = requirements - repo.supportedformats

        if missingreqs:

            raise error.UnsupportedBundleSpecification(

                    _('missing support for repository features: %s') %

                      ', '.join(sorted(missingreqs)))

    if not externalnames:

        engine = util.compengines.forbundlename(compression)

        compression = engine.bundletype()[1]

        version = _bundlespeccgversions[version]

    return compression, version, params

def readbundle(ui, fh, fname, vfs=None):

    header = changegroup.readexactly(fh, 4)

    alg = None

    if not fname:

        fname = "stream"

        if not header.startswith('HG') and header.startswith('\0'):

            fh = changegroup.headerlessfixup(fh, header)

            header = "HG10"

            alg = 'UN'

    elif vfs:

        fname = vfs.join(fname)

    magic, version = header[0:2], header[2:4]

    if magic != 'HG':

        raise error.Abort(_('%s: not a Mercurial bundle') % fname)

    if version == '10':

        if alg is None:

            alg = changegroup.readexactly(fh, 2)

        return changegroup.cg1unpacker(fh, alg)

    elif version.startswith('2'):

        return bundle2.getunbundler(ui, fh, magicstring=magic + version)

    elif version == 'S1':

        return streamclone.streamcloneapplier(fh)

    else:

        raise error.Abort(_('%s: unknown bundle version %s') % (fname, version))

def getbundlespec(ui, fh):

    """Infer the bundlespec from a bundle file handle.

    The input file handle is seeked and the original seek position is not

    restored.

    """

    def speccompression(alg):

        try:

            return util.compengines.forbundletype(alg).bundletype()[0]

        except KeyError:

            return None

    b = readbundle(ui, fh, None)

    if isinstance(b, changegroup.cg1unpacker):

        alg = b._type

        if alg == '_truncatedBZ':

            alg = 'BZ'

        comp = speccompression(alg)

        if not comp:

            raise error.Abort(_('unknown compression algorithm: %s') % alg)

        return '%s-v1' % comp

    elif isinstance(b, bundle2.unbundle20):

        if 'Compression' in b.params:

            comp = speccompression(b.params['Compression'])

            if not comp:

                raise error.Abort(_('unknown compression algorithm: %s') % comp)

        else:

            comp = 'none'

        version = None

        for part in b.iterparts():

            if part.type == 'changegroup':

                version = part.params['version']

                if version in ('01', '02'):

                    version = 'v2'

                else:

                    raise error.Abort(_('changegroup version %s does not have '

                                        'a known bundlespec') % version,

                                      hint=_('try upgrading your Mercurial '

                                              'client'))

        if not version:

            raise error.Abort(_('could not identify changegroup version in '

                                'bundle'))

        return '%s-%s' % (comp, version)

    elif isinstance(b, streamclone.streamcloneapplier):

        requirements = streamclone.readbundle1header(fh)[2]

        params = 'requirements=%s' % ','.join(sorted(requirements))

        return 'none-packed1;%s' % urlreq.quote(params)

    else:

        raise error.Abort(_('unknown bundle type: %s') % b)

def _computeoutgoing(repo, heads, common):

    """Computes which revs are outgoing given a set of common

    and a set of heads.

    This is a separate function so extensions can have access to

    the logic.

    Returns a discovery.outgoing object.

    """

    cl = repo.changelog

    if common:

        hasnode = cl.hasnode

        common = [n for n in common if hasnode(n)]

    else:

        common = [nullid]

    if not heads:

        heads = cl.heads()

    return discovery.outgoing(repo, common, heads)

def _forcebundle1(op):

    """return true if a pull/push must use bundle1

    This function is used to allow testing of the older bundle version"""

    ui = op.repo.ui

    forcebundle1 = False

    # The goal is this config is to allow developer to choose the bundle

    # version used during exchanged. This is especially handy during test.

    # Value is a list of bundle version to be picked from, highest version

    # should be used.

    #

    # developer config: devel.legacy.exchange

    exchange = ui.configlist('devel', 'legacy.exchange')

    forcebundle1 = 'bundle2' not in exchange and 'bundle1' in exchange

    return forcebundle1 or not op.remote.capable('bundle2')

class pushoperation(object):

    """A object that represent a single push operation

    Its purpose is to carry push related state and very common operations.

    A new pushoperation should be created at the beginning of each push and

    discarded afterward.

    """

    def __init__(self, repo, remote, force=False, revs=None, newbranch=False,

                 bookmarks=()):

        # repo we push from

        self.repo = repo

        self.ui = repo.ui

        # repo we push to

        self.remote = remote

        # force option provided

        self.force = force

        # revs to be pushed (None is "all")

        self.revs = revs

        # bookmark explicitly pushed

        self.bookmarks = bookmarks

        # allow push of new branch

        self.newbranch = newbranch

        # did a local lock get acquired?

        self.locallocked = None

        # step already performed

        # (used to check what steps have been already performed through bundle2)

        self.stepsdone = set()

        # Integer version of the changegroup push result

        # - None means nothing to push

        # - 0 means HTTP error

        # - 1 means we pushed and remote head count is unchanged *or*

        #   we have outgoing changesets but refused to push

        # - other values as described by addchangegroup()

        self.cgresult = None

        # Boolean value for the bookmark push

        self.bkresult = None

        # discover.outgoing object (contains common and outgoing data)

        self.outgoing = None

        # all remote topological heads before the push

        self.remoteheads = None

        # Details of the remote branch pre and post push

        #

        # mapping: {'branch': ([remoteheads],

        #                      [newheads],

        #                      [unsyncedheads],

        #                      [discardedheads])}

        # - branch: the branch name

        # - remoteheads: the list of remote heads known locally

        #                None if the branch is new

        # - newheads: the new remote heads (known locally) with outgoing pushed

        # - unsyncedheads: the list of remote heads unknown locally.

        # - discardedheads: the list of remote heads made obsolete by the push

        self.pushbranchmap = None

        # testable as a boolean indicating if any nodes are missing locally.

        self.incoming = None

        # phases changes that must be pushed along side the changesets

        self.outdatedphases = None

        # phases changes that must be pushed if changeset push fails

        self.fallbackoutdatedphases = None

        # outgoing obsmarkers

        self.outobsmarkers = set()

        # outgoing bookmarks

        self.outbookmarks = []

        # transaction manager

        self.trmanager = None

        # map { pushkey partid -> callback handling failure}

        # used to handle exception from mandatory pushkey part failure

        self.pkfailcb = {}

    @util.propertycache

    def futureheads(self):

        """future remote heads if the changeset push succeeds"""

        return self.outgoing.missingheads

    @util.propertycache

    def fallbackheads(self):

        """future remote heads if the changeset push fails"""

        if self.revs is None:

            # not target to push, all common are relevant

            return self.outgoing.commonheads

        unfi = self.repo.unfiltered()

        # I want cheads = heads(::missingheads and ::commonheads)

        # (missingheads is revs with secret changeset filtered out)

        #

        # This can be expressed as:

        #     cheads = ( (missingheads and ::commonheads)

        #              + (commonheads and ::missingheads))"

        #              )

        #

        # while trying to push we already computed the following:

        #     common = (::commonheads)

        #     missing = ((commonheads::missingheads) - commonheads)

        #

        # We can pick:

        # * missingheads part of common (::commonheads)

        common = self.outgoing.common

        nm = self.repo.changelog.nodemap

        cheads = [node for node in self.revs if nm[node] in common]

        # and

        # * commonheads parents on missing

        revset = unfi.set('%ln and parents(roots(%ln))',

                         self.outgoing.commonheads,

                         self.outgoing.missing)

        cheads.extend(c.node() for c in revset)

        return cheads

    @property

    def commonheads(self):

        """set of all common heads after changeset bundle push"""

        if self.cgresult:

            return self.futureheads

        else:

            return self.fallbackheads

# mapping of message used when pushing bookmark

bookmsgmap = {'update': (_("updating bookmark %s\n"),

                         _('updating bookmark %s failed!\n')),

              'export': (_("exporting bookmark %s\n"),

                         _('exporting bookmark %s failed!\n')),

              'delete': (_("deleting remote bookmark %s\n"),

                         _('deleting remote bookmark %s failed!\n')),

              }

def push(repo, remote, force=False, revs=None, newbranch=False, bookmarks=(),

         opargs=None):

    '''Push outgoing changesets (limited by revs) from a local

    repository to remote. Return an integer:

      - None means nothing to push

      - 0 means HTTP error

      - 1 means we pushed and remote head count is unchanged *or*

        we have outgoing changesets but refused to push

      - other values as described by addchangegroup()

    '''

    if opargs is None:

        opargs = {}

    pushop = pushoperation(repo, remote, force, revs, newbranch, bookmarks,

                           **opargs)

    if pushop.remote.local():

        missing = (set(pushop.repo.requirements)

                   - pushop.remote.local().supported)

        if missing:

            msg = _("required features are not"

                    " supported in the destination:"

                    " %s") % (', '.join(sorted(missing)))

            raise error.Abort(msg)

    # there are two ways to push to remote repo:

    #

    # addchangegroup assumes local user can lock remote

    # repo (local filesystem, old ssh servers).

    #

    # unbundle assumes local user cannot lock remote repo (new ssh

    # servers, http servers).

    if not pushop.remote.canpush():

        raise error.Abort(_("destination does not support push"))

    # get local lock as we might write phase data

    localwlock = locallock = None

    try:

        # bundle2 push may receive a reply bundle touching bookmarks or other

        # things requiring the wlock. Take it now to ensure proper ordering.

        maypushback = pushop.ui.configbool('experimental', 'bundle2.pushback')

        if (not _forcebundle1(pushop)) and maypushback:

            localwlock = pushop.repo.wlock()

        locallock = pushop.repo.lock()

        pushop.locallocked = True

    except IOError as err:

        pushop.locallocked = False

        if err.errno != errno.EACCES:

            raise

        # source repo cannot be locked.

        # We do not abort the push, but just disable the local phase

        # synchronisation.

        msg = 'cannot lock source repository: %s\n' % err

        pushop.ui.debug(msg)

    try:

        if pushop.locallocked:

            pushop.trmanager = transactionmanager(pushop.repo,

                                                  'push-response',

                                                  pushop.remote.url())

        pushop.repo.checkpush(pushop)

        lock = None

        unbundle = pushop.remote.capable('unbundle')

        if not unbundle:

            lock = pushop.remote.lock()

        try:

            _pushdiscovery(pushop)

            if not _forcebundle1(pushop):

                _pushbundle2(pushop)

            _pushchangeset(pushop)

            _pushsyncphase(pushop)

            _pushobsolete(pushop)

            _pushbookmark(pushop)

        finally:

            if lock is not None:

                lock.release()

        if pushop.trmanager:

            pushop.trmanager.close()