# changegroup.py - Mercurial changegroup manipulation functions

#

#  Copyright 2006 Olivia Mackall <olivia@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.

import os

import struct

import weakref

from .i18n import _

from .node import (

    hex,

    nullrev,

    short,

)

from .pycompat import open

from . import (

    error,

    match as matchmod,

    mdiff,

    phases,

    pycompat,

    requirements,

    scmutil,

    util,

)

from .interfaces import repository

from .revlogutils import sidedata as sidedatamod

from .revlogutils import constants as revlog_constants

from .utils import storageutil

_CHANGEGROUPV1_DELTA_HEADER = struct.Struct(b"20s20s20s20s")

_CHANGEGROUPV2_DELTA_HEADER = struct.Struct(b"20s20s20s20s20s")

_CHANGEGROUPV3_DELTA_HEADER = struct.Struct(b">20s20s20s20s20sH")

_CHANGEGROUPV4_DELTA_HEADER = struct.Struct(b">B20s20s20s20s20sH")

LFS_REQUIREMENT = b'lfs'

readexactly = util.readexactly

def getchunk(stream):

    """return the next chunk from stream as a string"""

    d = readexactly(stream, 4)

    l = struct.unpack(b">l", d)[0]

    if l <= 4:

        if l:

            raise error.Abort(_(b"invalid chunk length %d") % l)

        return b""

    return readexactly(stream, l - 4)

def chunkheader(length):

    """return a changegroup chunk header (string)"""

    return struct.pack(b">l", length + 4)

def closechunk():

    """return a changegroup chunk header (string) for a zero-length chunk"""

    return struct.pack(b">l", 0)

def _fileheader(path):

    """Obtain a changegroup chunk header for a named path."""

    return chunkheader(len(path)) + path

def writechunks(ui, chunks, filename, vfs=None):

    """Write chunks to a file and return its filename.

    The stream is assumed to be a bundle file.

    Existing files will not be overwritten.

    If no filename is specified, a temporary file is created.

    """

    fh = None

    cleanup = None

    try:

        if filename:

            if vfs:

                fh = vfs.open(filename, b"wb")

            else:

                # Increase default buffer size because default is usually

                # small (4k is common on Linux).

                fh = open(filename, b"wb", 131072)

        else:

            fd, filename = pycompat.mkstemp(prefix=b"hg-bundle-", suffix=b".hg")

            fh = os.fdopen(fd, "wb")

        cleanup = filename

        for c in chunks:

            fh.write(c)

        cleanup = None

        return filename

    finally:

        if fh is not None:

            fh.close()

        if cleanup is not None:

            if filename and vfs:

                vfs.unlink(cleanup)

            else:

                os.unlink(cleanup)

def _dbg_ubdl_line(

    ui,

    indent,

    key,

    base_value=None,

    percentage_base=None,

    percentage_key=None,

):

    """Print one line of debug_unbundle_debug_info"""

    line = b"DEBUG-UNBUNDLING: "

    line += b' ' * (2 * indent)

    key += b":"

    padding = b''

    if base_value is not None:

        assert len(key) + 1 + (2 * indent) <= _KEY_PART_WIDTH

        line += key.ljust(_KEY_PART_WIDTH - (2 * indent))

        if isinstance(base_value, float):

            line += b"%14.3f seconds" % base_value

        else:

            line += b"%10d" % base_value

            padding = b'            '

    else:

        line += key

    if percentage_base is not None:

        line += padding

        padding = b''

        assert base_value is not None

        percentage = base_value * 100 // percentage_base

        if percentage_key is not None:

            line += b" (%3d%% of %s)" % (

                percentage,

                percentage_key,

            )

        else:

            line += b" (%3d%%)" % percentage

    line += b'\n'

    ui.write_err(line)

def _sumf(items):

    # python < 3.8 does not support a `start=0.0` argument to sum

    # So we have to cheat a bit until we drop support for those version

    if not items:

        return 0.0

    return sum(items)

def display_unbundle_debug_info(ui, debug_info):

    """display an unbundling report from debug information"""

    cl_info = []

    mn_info = []

    fl_info = []

    _dispatch = [

        (b'CHANGELOG:', cl_info),

        (b'MANIFESTLOG:', mn_info),

        (b'FILELOG:', fl_info),

    ]

    for e in debug_info:

        for prefix, info in _dispatch:

            if e["target-revlog"].startswith(prefix):

                info.append(e)

                break

        else:

            assert False, 'unreachable'

    each_info = [

        (b'changelog', cl_info),

        (b'manifests', mn_info),

        (b'files', fl_info),

    ]

    # General Revision Countss

    _dbg_ubdl_line(ui, 0, b'revisions', len(debug_info))

    for key, info in each_info:

        if not info:

            continue

        _dbg_ubdl_line(ui, 1, key, len(info), len(debug_info))

    # General Time spent

    all_durations = [e['duration'] for e in debug_info]

    all_durations.sort()

    total_duration = _sumf(all_durations)

    _dbg_ubdl_line(ui, 0, b'total-time', total_duration)

    for key, info in each_info:

        if not info:

            continue

        durations = [e['duration'] for e in info]

        durations.sort()

        _dbg_ubdl_line(ui, 1, key, _sumf(durations), total_duration)

    # Count and cache reuse per delta types

    each_types = {}

    for key, info in each_info:

        each_types[key] = types = {

            b'full': 0,

            b'full-cached': 0,

            b'snapshot': 0,

            b'snapshot-cached': 0,

            b'delta': 0,

            b'delta-cached': 0,

            b'unknown': 0,

            b'unknown-cached': 0,

        }

        for e in info:

            types[e['type']] += 1

            if e['using-cached-base']:

                types[e['type'] + b'-cached'] += 1

    EXPECTED_TYPES = (b'full', b'snapshot', b'delta', b'unknown')

    if debug_info:

        _dbg_ubdl_line(ui, 0, b'type-count')

    for key, info in each_info:

        if info:

            _dbg_ubdl_line(ui, 1, key)

        t = each_types[key]

        for tn in EXPECTED_TYPES:

            if t[tn]:

                tc = tn + b'-cached'

                _dbg_ubdl_line(ui, 2, tn, t[tn])

                _dbg_ubdl_line(ui, 3, b'cached', t[tc], t[tn])

    # time perf delta types and reuse

    each_type_time = {}

    for key, info in each_info:

        each_type_time[key] = t = {

            b'full': [],

            b'full-cached': [],

            b'snapshot': [],

            b'snapshot-cached': [],

            b'delta': [],

            b'delta-cached': [],

            b'unknown': [],

            b'unknown-cached': [],

        }

        for e in info:

            t[e['type']].append(e['duration'])

            if e['using-cached-base']:

                t[e['type'] + b'-cached'].append(e['duration'])

        for t_key, value in list(t.items()):

            value.sort()

            t[t_key] = _sumf(value)

    if debug_info:

        _dbg_ubdl_line(ui, 0, b'type-time')

    for key, info in each_info:

        if info:

            _dbg_ubdl_line(ui, 1, key)

        t = each_type_time[key]

        td = total_duration  # to same space on next lines

        for tn in EXPECTED_TYPES:

            if t[tn]:

                tc = tn + b'-cached'

                _dbg_ubdl_line(ui, 2, tn, t[tn], td, b"total")

                _dbg_ubdl_line(ui, 3, b'cached', t[tc], td, b"total")

class cg1unpacker:

    """Unpacker for cg1 changegroup streams.

    A changegroup unpacker handles the framing of the revision data in

    the wire format. Most consumers will want to use the apply()

    method to add the changes from the changegroup to a repository.

    If you're forwarding a changegroup unmodified to another consumer,

    use getchunks(), which returns an iterator of changegroup

    chunks. This is mostly useful for cases where you need to know the

    data stream has ended by observing the end of the changegroup.

    deltachunk() is useful only if you're applying delta data. Most

    consumers should prefer apply() instead.

    A few other public methods exist. Those are used only for

    bundlerepo and some debug commands - their use is discouraged.

    """

    deltaheader = _CHANGEGROUPV1_DELTA_HEADER

    deltaheadersize = deltaheader.size

    version = b'01'

    _grouplistcount = 1  # One list of files after the manifests

    def __init__(self, fh, alg, extras=None):

        if alg is None:

            alg = b'UN'

        if alg not in util.compengines.supportedbundletypes:

            raise error.Abort(_(b'unknown stream compression type: %s') % alg)

        if alg == b'BZ':

            alg = b'_truncatedBZ'

        compengine = util.compengines.forbundletype(alg)

        self._stream = compengine.decompressorreader(fh)

        self._type = alg

        self.extras = extras or {}

        self.callback = None

    # These methods (compressed, read, seek, tell) all appear to only

    # be used by bundlerepo, but it's a little hard to tell.

    def compressed(self):

        return self._type is not None and self._type != b'UN'

    def read(self, l):

        return self._stream.read(l)

    def seek(self, pos):

        return self._stream.seek(pos)

    def tell(self):

        return self._stream.tell()

    def close(self):

        return self._stream.close()

    def _chunklength(self):

        d = readexactly(self._stream, 4)

        l = struct.unpack(b">l", d)[0]

        if l <= 4:

            if l:

                raise error.Abort(_(b"invalid chunk length %d") % l)

            return 0

        if self.callback:

            self.callback()

        return l - 4

    def changelogheader(self):

        """v10 does not have a changelog header chunk"""

        return {}

    def manifestheader(self):

        """v10 does not have a manifest header chunk"""

        return {}

    def filelogheader(self):

        """return the header of the filelogs chunk, v10 only has the filename"""

        l = self._chunklength()

        if not l:

            return {}

        fname = readexactly(self._stream, l)

        return {b'filename': fname}

    def _deltaheader(self, headertuple, prevnode):

        node, p1, p2, cs = headertuple

        if prevnode is None:

            deltabase = p1

        else:

            deltabase = prevnode

        flags = 0

        protocol_flags = 0

        return node, p1, p2, deltabase, cs, flags, protocol_flags

    def deltachunk(self, prevnode):

        # Chunkdata: (node, p1, p2, cs, deltabase, delta, flags, sidedata, proto_flags)

        l = self._chunklength()

        if not l:

            return {}

        headerdata = readexactly(self._stream, self.deltaheadersize)

        header = self.deltaheader.unpack(headerdata)

        delta = readexactly(self._stream, l - self.deltaheadersize)

        header = self._deltaheader(header, prevnode)

        node, p1, p2, deltabase, cs, flags, protocol_flags = header

        return node, p1, p2, cs, deltabase, delta, flags, {}, protocol_flags

    def getchunks(self):

        """returns all the chunks contains in the bundle

        Used when you need to forward the binary stream to a file or another

        network API. To do so, it parse the changegroup data, otherwise it will

        block in case of sshrepo because it don't know the end of the stream.

        """

        # For changegroup 1 and 2, we expect 3 parts: changelog, manifestlog,

        # and a list of filelogs. For changegroup 3, we expect 4 parts:

        # changelog, manifestlog, a list of tree manifestlogs, and a list of

        # filelogs.

        #

        # Changelog and manifestlog parts are terminated with empty chunks. The

        # tree and file parts are a list of entry sections. Each entry section

        # is a series of chunks terminating in an empty chunk. The list of these