# branchmap.py - logic to computes, maintain and stores branchmap for local repo

#

# Copyright 2005-2007 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 struct

from .node import (

    bin,

    hex,

    nullrev,

)

from typing import (

    Callable,

    Dict,

    Iterable,

    List,

    Optional,

    Set,

    TYPE_CHECKING,

    Tuple,

    Union,

)

from . import (

    encoding,

    error,

    obsolete,

    scmutil,

    util,

)

from .utils import (

    repoviewutil,

    stringutil,

)

if TYPE_CHECKING:

    from . import localrepo

    assert [localrepo]

subsettable = repoviewutil.subsettable

calcsize = struct.calcsize

pack_into = struct.pack_into

unpack_from = struct.unpack_from

class BranchMapCache:

    """mapping of filtered views of repo with their branchcache"""

    def __init__(self):

        self._per_filter = {}

    def __getitem__(self, repo):

        self.updatecache(repo)

        return self._per_filter[repo.filtername]

    def updatecache(self, repo):

        """Update the cache for the given filtered view on a repository"""

        # This can trigger updates for the caches for subsets of the filtered

        # view, e.g. when there is no cache for this filtered view or the cache

        # is stale.

        cl = repo.changelog

        filtername = repo.filtername

        bcache = self._per_filter.get(filtername)

        if bcache is None or not bcache.validfor(repo):

            # cache object missing or cache object stale? Read from disk

            bcache = branchcache.fromfile(repo)

        revs = []

        if bcache is None:

            # no (fresh) cache available anymore, perhaps we can re-use

            # the cache for a subset, then extend that to add info on missing

            # revisions.

            subsetname = subsettable.get(filtername)

            if subsetname is not None:

                subset = repo.filtered(subsetname)

                bcache = self[subset].copy()

                extrarevs = subset.changelog.filteredrevs - cl.filteredrevs

                revs.extend(r for r in extrarevs if r <= bcache.tiprev)

            else:

                # nothing to fall back on, start empty.

                bcache = branchcache(repo)

        revs.extend(cl.revs(start=bcache.tiprev + 1))

        if revs:

            bcache.update(repo, revs)

        assert bcache.validfor(repo), filtername

        self._per_filter[repo.filtername] = bcache

    def replace(self, repo, remotebranchmap):

        """Replace the branchmap cache for a repo with a branch mapping.

        This is likely only called during clone with a branch map from a

        remote.

        """

        cl = repo.changelog

        clrev = cl.rev

        clbranchinfo = cl.branchinfo

        rbheads = []

        closed = set()

        for bheads in remotebranchmap.values():

            rbheads += bheads

            for h in bheads:

                r = clrev(h)

                b, c = clbranchinfo(r)

                if c:

                    closed.add(h)

        if rbheads:

            rtiprev = max((int(clrev(node)) for node in rbheads))

            cache = branchcache(

                repo,

                remotebranchmap,

                repo[rtiprev].node(),

                rtiprev,

                closednodes=closed,

            )

            # Try to stick it as low as possible

            # filter above served are unlikely to be fetch from a clone

            for candidate in (b'base', b'immutable', b'served'):

                rview = repo.filtered(candidate)

                if cache.validfor(rview):

                    self._per_filter[candidate] = cache

                    cache.write(rview)

                    return

    def clear(self):

        self._per_filter.clear()

    def write_delayed(self, repo):

        unfi = repo.unfiltered()

        for filtername, cache in self._per_filter.items():

            if cache._delayed:

                repo = unfi.filtered(filtername)

                cache.write(repo)

def _unknownnode(node):

    """raises ValueError when branchcache found a node which does not exists"""

    raise ValueError('node %s does not exist' % node.hex())

def _branchcachedesc(repo):

    if repo.filtername is not None:

        return b'branch cache (%s)' % repo.filtername

    else:

        return b'branch cache'

class branchcache:

    """A dict like object that hold branches heads cache.

    This cache is used to avoid costly computations to determine all the

    branch heads of a repo.

    The cache is serialized on disk in the following format:

    <tip hex node> <tip rev number> [optional filtered repo hex hash]

    <branch head hex node> <open/closed state> <branch name>

    <branch head hex node> <open/closed state> <branch name>

    ...

    The first line is used to check if the cache is still valid. If the

    branch cache is for a filtered repo view, an optional third hash is

    included that hashes the hashes of all filtered and obsolete revisions.

    The open/closed state is represented by a single letter 'o' or 'c'.

    This field can be used to avoid changelog reads when determining if a

    branch head closes a branch or not.

    """

    def __init__(

        self,

        repo: "localrepo.localrepository",

        entries: Union[

            Dict[bytes, List[bytes]], Iterable[Tuple[bytes, List[bytes]]]

        ] = (),

        tipnode: Optional[bytes] = None,

        tiprev: Optional[int] = nullrev,

        filteredhash: Optional[bytes] = None,

        closednodes: Optional[Set[bytes]] = None,

        hasnode: Optional[Callable[[bytes], bool]] = None,

    ) -> None:

        """hasnode is a function which can be used to verify whether changelog

        has a given node or not. If it's not provided, we assume that every node

        we have exists in changelog"""

        self._repo = repo

        self._delayed = False

        if tipnode is None:

            self.tipnode = repo.nullid

        else:

            self.tipnode = tipnode

        self.tiprev = tiprev

        self.filteredhash = filteredhash

        # closednodes is a set of nodes that close their branch. If the branch

        # cache has been updated, it may contain nodes that are no longer

        # heads.

        if closednodes is None:

            self._closednodes = set()

        else:

            self._closednodes = closednodes

        self._entries = dict(entries)

        # whether closed nodes are verified or not

        self._closedverified = False

        # branches for which nodes are verified

        self._verifiedbranches = set()

        self._hasnode = hasnode

        if self._hasnode is None:

            self._hasnode = lambda x: True

    def _verifyclosed(self):

        """verify the closed nodes we have"""

        if self._closedverified:

            return

        for node in self._closednodes:

            if not self._hasnode(node):

                _unknownnode(node)

        self._closedverified = True

    def _verifybranch(self, branch):

        """verify head nodes for the given branch."""

        if branch not in self._entries or branch in self._verifiedbranches:

            return

        for n in self._entries[branch]:

            if not self._hasnode(n):

                _unknownnode(n)

        self._verifiedbranches.add(branch)

    def _verifyall(self):

        """verifies nodes of all the branches"""

        needverification = set(self._entries.keys()) - self._verifiedbranches

        for b in needverification:

            self._verifybranch(b)

    def __iter__(self):

        return iter(self._entries)

    def __setitem__(self, key, value):

        self._entries[key] = value

    def __getitem__(self, key):

        self._verifybranch(key)

        return self._entries[key]

    def __contains__(self, key):

        self._verifybranch(key)

        return key in self._entries

    def iteritems(self):

        for k, v in self._entries.items():

            self._verifybranch(k)

            yield k, v

    items = iteritems

    def hasbranch(self, label):

        """checks whether a branch of this name exists or not"""

        self._verifybranch(label)

        return label in self._entries

    @classmethod

    def fromfile(cls, repo):

        f = None

        try:

            f = repo.cachevfs(cls._filename(repo))

            lineiter = iter(f)

            cachekey = next(lineiter).rstrip(b'\n').split(b" ", 2)

            last, lrev = cachekey[:2]

            last, lrev = bin(last), int(lrev)

            filteredhash = None

            hasnode = repo.changelog.hasnode

            if len(cachekey) > 2:

                filteredhash = bin(cachekey[2])

            bcache = cls(

                repo,

                tipnode=last,

                tiprev=lrev,

                filteredhash=filteredhash,

                hasnode=hasnode,

            )

            if not bcache.validfor(repo):

                # invalidate the cache

                raise ValueError('tip differs')

            bcache.load(repo, lineiter)

        except (IOError, OSError):

            return None

        except Exception as inst:

            if repo.ui.debugflag:

                msg = b'invalid %s: %s\n'

                repo.ui.debug(

                    msg

                    % (

                        _branchcachedesc(repo),

                        stringutil.forcebytestr(inst),

                    )

                )

            bcache = None

        finally:

            if f:

                f.close()

        return bcache

    def load(self, repo, lineiter):

        """fully loads the branchcache by reading from the file using the line

        iterator passed"""

        for line in lineiter:

            line = line.rstrip(b'\n')

            if not line:

                continue

            node, state, label = line.split(b" ", 2)

            if state not in b'oc':

                raise ValueError('invalid branch state')

            label = encoding.tolocal(label.strip())

            node = bin(node)

            self._entries.setdefault(label, []).append(node)

            if state == b'c':

                self._closednodes.add(node)

    @staticmethod

    def _filename(repo):

        """name of a branchcache file for a given repo or repoview"""

        filename = b"branch2"

        if repo.filtername:

            filename = b'%s-%s' % (filename, repo.filtername)

        return filename

    def validfor(self, repo):

        """check that cache contents are valid for (a subset of) this repo

        - False when the order of changesets changed or if we detect a strip.

        - True when cache is up-to-date for the current repo or its subset."""

        try:

            node = repo.changelog.node(self.tiprev)

        except IndexError:

            # changesets were stripped and now we don't even have enough to

            # find tiprev

            return False

        if self.tipnode != node:

            # tiprev doesn't correspond to tipnode: repo was stripped, or this

            # repo has a different order of changesets

            return False

        tiphash = scmutil.filteredhash(repo, self.tiprev, needobsolete=True)

        # hashes don't match if this repo view has a different set of filtered

        # revisions (e.g. due to phase changes) or obsolete revisions (e.g.

        # history was rewritten)

        return self.filteredhash == tiphash

    def _branchtip(self, heads):

        """Return tuple with last open head in heads and false,

        otherwise return last closed head and true."""

        tip = heads[-1]

        closed = True

        for h in reversed(heads):

            if h not in self._closednodes:

                tip = h

                closed = False

                break

        return tip, closed

    def branchtip(self, branch):

        """Return the tipmost open head on branch head, otherwise return the

        tipmost closed head on branch.

        Raise KeyError for unknown branch."""

        return self._branchtip(self[branch])[0]

    def iteropen(self, nodes):

        return (n for n in nodes if n not in self._closednodes)

    def branchheads(self, branch, closed=False):

        self._verifybranch(branch)

        heads = self._entries[branch]

        if not closed: