# 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 i18n import _

from node import hex, nullid

import errno, urllib, urllib2

import util, scmutil, changegroup, base85, error

import discovery, phases, obsolete, bookmarks as bookmod, bundle2, pushkey

import lock as lockmod

import streamclone

import sslutil

import tags

import url as urlmod

# Maps bundle compression human names to internal representation.

_bundlespeccompressions = {'none': None,

                           'bzip2': 'BZ',

                           'gzip': 'GZ',

                          }

# Maps bundle version human names to changegroup versions.

_bundlespeccgversions = {'v1': '01',

                         'v2': '02',

                         'packed1': 's1',

                         'bundle2': '02', #legacy

                        }

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 interpretted 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 = urllib.unquote(key)

            value = urllib.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 _bundlespeccompressions:

            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 _bundlespeccompressions:

            compression = spec

            version = 'v1'

            if 'generaldelta' in repo.requirements:

                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)

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

        compression = _bundlespeccompressions[compression]

        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 buildobsmarkerspart(bundler, markers):

    """add an obsmarker part to the bundler with <markers>

    No part is created if markers is empty.

    Raises ValueError if the bundler doesn't support any known obsmarker format.

    """

    if markers:

        remoteversions = bundle2.obsmarkersversion(bundler.capabilities)

        version = obsolete.commonversion(remoteversions)

        if version is None:

            raise ValueError('bundler do not support common obsmarker format')

        stream = obsolete.encodemarkers(markers, True, version=version)

        return bundler.newpart('obsmarkers', data=stream)

    return None

def _canusebundle2(op):

    """return true if a pull/push can use bundle2

    Feel free to nuke this function when we drop the experimental option"""

    return (op.repo.ui.configbool('experimental', 'bundle2-exp', True)

            and op.remote.capable('bundle2'))

class pushoperation(object):

    """A object that represent a single push operation

    It purpose is to carry push related state and very common operation.

    A new 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 heads before the push

        self.remoteheads = 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 _canusebundle2(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 _canusebundle2(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()

    finally:

        if pushop.trmanager:

            pushop.trmanager.release()

        if locallock is not None:

            locallock.release()

        if localwlock is not None:

            localwlock.release()

    return pushop

# list of steps to perform discovery before push

pushdiscoveryorder = []