# 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 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',

                         'bundle2': '02', #legacy

                        }

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

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

    Where <compression> is one of the supported compression formats

    and <type> is (currently) a version string.

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

    it is optional.

    Returns a 2-tuple of (compression, version). 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.

    """

    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)

        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

        if spec in _bundlespeccompressions:

            compression = spec

            version = 'v1'

            if 'generaldelta' in repo.requirements:

                version = 'v2'

        elif spec in _bundlespeccgversions:

            compression = 'bzip2'

            version = spec

        else:

            raise error.UnsupportedBundleSpecification(

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

    compression = _bundlespeccompressions[compression]

    version = _bundlespeccgversions[version]

    return compression, version

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)

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

    '''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()

    '''

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

    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(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 = []

# Mapping between step name and function

#

# This exists to help extensions wrap steps if necessary

pushdiscoverymapping = {}

def pushdiscovery(stepname):

    """decorator for function performing discovery before push

    The function is added to the step -> function mapping and appended to the

    list of steps.  Beware that decorated function will be added in order (this

    may matter).

    You can only use this decorator for a new step, if you want to wrap a step

    from an extension, change the pushdiscovery dictionary directly."""

    def dec(func):

        assert stepname not in pushdiscoverymapping

        pushdiscoverymapping[stepname] = func

        pushdiscoveryorder.append(stepname)

        return func

    return dec

def _pushdiscovery(pushop):

    """Run all discovery steps"""

    for stepname in pushdiscoveryorder:

        step = pushdiscoverymapping[stepname]

        step(pushop)

@pushdiscovery('changeset')

def _pushdiscoverychangeset(pushop):

    """discover the changeset that need to be pushed"""

    fci = discovery.findcommonincoming

    commoninc = fci(pushop.repo, pushop.remote, force=pushop.force)

    common, inc, remoteheads = commoninc

    fco = discovery.findcommonoutgoing

    outgoing = fco(pushop.repo, pushop.remote, onlyheads=pushop.revs,

                   commoninc=commoninc, force=pushop.force)

    pushop.outgoing = outgoing

    pushop.remoteheads = remoteheads

    pushop.incoming = inc

@pushdiscovery('phase')

def _pushdiscoveryphase(pushop):

    """discover the phase that needs to be pushed

    (computed for both success and failure case for changesets push)"""

    outgoing = pushop.outgoing

    unfi = pushop.repo.unfiltered()

    remotephases = pushop.remote.listkeys('phases')

    publishing = remotephases.get('publishing', False)

    if (pushop.ui.configbool('ui', '_usedassubrepo', False)

        and remotephases    # server supports phases

        and not pushop.outgoing.missing # no changesets to be pushed

        and publishing):