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

    Any,

    Callable,

    Dict,

    Iterable,

    List,

    Optional,

    Set,

    TYPE_CHECKING,

    Tuple,

    Union,

    cast,

)

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)

        bcache = self._per_filter[repo.filtername]

        assert bcache._filtername == repo.filtername, (

            bcache._filtername,

            repo.filtername,

        )

        return bcache

    def update_disk(self, repo):

        """ensure and up-to-date cache is (or will be) written on disk

        The cache for this repository view is updated  if needed and written on

        disk.

        If a transaction is in progress, the writing is schedule to transaction

        close. See the `BranchMapCache.write_dirty` method.

        This method exist independently of __getitem__ as it is sometime useful

        to signal that we have no intend to use the data in memory yet.

        """

        self.updatecache(repo)

        bcache = self._per_filter[repo.filtername]

        assert bcache._filtername == repo.filtername, (

            bcache._filtername,

            repo.filtername,

        )

        tr = repo.currenttransaction()

        if getattr(tr, 'finalized', True):

            bcache.sync_disk(repo)

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

                self.updatecache(subset)

                bcache = self._per_filter[subset.filtername].inherit_for(repo)

                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 = new_branch_cache(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 = new_branch_cache(

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

                    cache._filtername = candidate

                    self._per_filter[candidate] = cache

                    cache._state = STATE_DIRTY

                    cache.write(rview)

                    return

    def clear(self):

        self._per_filter.clear()

    def write_dirty(self, repo):

        unfi = repo.unfiltered()

        for filtername in repoviewutil.get_ordered_subset():

            cache = self._per_filter.get(filtername)

            if cache is None:

                continue

            if filtername is None:

                repo = unfi

            else:

                repo = unfi.filtered(filtername)

            cache.sync_disk(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 _BaseBranchCache:

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

    """

    def __init__(

        self,

        repo: "localrepo.localrepository",

        entries: Union[

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

        ] = (),

        closed_nodes: Optional[Set[bytes]] = 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"""

        # 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 closed_nodes is None:

            closed_nodes = set()

        self._closednodes = set(closed_nodes)

        self._entries = dict(entries)

    def __iter__(self):

        return iter(self._entries)

    def __setitem__(self, key, value):

        self._entries[key] = value

    def __getitem__(self, key):

        return self._entries[key]

    def __contains__(self, key):

        return key in self._entries

    def iteritems(self):

        return self._entries.items()

    items = iteritems

    def hasbranch(self, label):

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

        return label in self._entries

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

        heads = self._entries[branch]

        if not closed:

            heads = list(self.iteropen(heads))

        return heads

    def iterbranches(self):

        for bn, heads in self.items():

            yield (bn, heads) + self._branchtip(heads)

    def iterheads(self):

        """returns all the heads"""

        return self._entries.values()

    def update(self, repo, revgen):

        """Given a branchhead cache, self, that may have extra nodes or be

        missing heads, and a generator of nodes that are strictly a superset of

        heads missing, this function updates self to be correct.

        """

        starttime = util.timer()

        cl = repo.changelog

        # Faster than using ctx.obsolete()

        obsrevs = obsolete.getrevs(repo, b'obsolete')

        # collect new branch entries

        newbranches = {}

        new_closed = set()

        obs_ignored = set()

        getbranchinfo = repo.revbranchcache().branchinfo

        max_rev = -1

        for r in revgen:

            max_rev = max(max_rev, r)

            if r in obsrevs:

                # We ignore obsolete changesets as they shouldn't be

                # considered heads.

                obs_ignored.add(r)

                continue

            branch, closesbranch = getbranchinfo(r)

            newbranches.setdefault(branch, []).append(r)

            if closesbranch:

                new_closed.add(r)

        if max_rev < 0:

            msg = "running branchcache.update without revision to update"

            raise error.ProgrammingError(msg)

        self._process_new(

            repo,

            newbranches,

            new_closed,

            obs_ignored,

            max_rev,

        )

        self._closednodes.update(cl.node(rev) for rev in new_closed)

        duration = util.timer() - starttime

        repo.ui.log(

            b'branchcache',

            b'updated %s in %.4f seconds\n',

            _branchcachedesc(repo),

            duration,

        )

        return max_rev

    def _process_new(

        self,

        repo,

        newbranches,

        new_closed,

        obs_ignored,

        max_rev,

    ):

        """update the branchmap from a set of new information"""

        # Delay fetching the topological heads until they are needed.

        # A repository without non-continous branches can skip this part.

        topoheads = None

        cl = repo.changelog

        getbranchinfo = repo.revbranchcache().branchinfo

        # Faster than using ctx.obsolete()

        obsrevs = obsolete.getrevs(repo, b'obsolete')

        # If a changeset is visible, its parents must be visible too, so

        # use the faster unfiltered parent accessor.

        parentrevs = cl._uncheckedparentrevs

        for branch, newheadrevs in newbranches.items():

            # For every branch, compute the new branchheads.

            # A branchhead is a revision such that no descendant is on

            # the same branch.

            #

            # The branchheads are computed iteratively in revision order.

            # This ensures topological order, i.e. parents are processed

            # before their children. Ancestors are inclusive here, i.e.

            # any revision is an ancestor of itself.

            #

            # Core observations:

            # - The current revision is always a branchhead for the

            #   repository up to that point.

            # - It is the first revision of the branch if and only if

            #   there was no branchhead before. In that case, it is the

            #   only branchhead as there are no possible ancestors on

            #   the same branch.

            # - If a parent is on the same branch, a branchhead can

            #   only be an ancestor of that parent, if it is parent

            #   itself. Otherwise it would have been removed as ancestor

            #   of that parent before.

            # - Therefore, if all parents are on the same branch, they

            #   can just be removed from the branchhead set.

            # - If one parent is on the same branch and the other is not

            #   and there was exactly one branchhead known, the existing

            #   branchhead can only be an ancestor if it is the parent.

            #   Otherwise it would have been removed as ancestor of

            #   the parent before. The other parent therefore can't have

            #   a branchhead as ancestor.

            # - In all other cases, the parents on different branches

            #   could have a branchhead as ancestor. Those parents are

            #   kept in the "uncertain" set. If all branchheads are also

            #   topological heads, they can't have descendants and further

            #   checks can be skipped. Otherwise, the ancestors of the

            #   "uncertain" set are removed from branchheads.

            #   This computation is heavy and avoided if at all possible.

            bheads = self._entries.get(branch, [])

            bheadset = {cl.rev(node) for node in bheads}

            uncertain = set()

            for newrev in sorted(newheadrevs):

                if not bheadset:

                    bheadset.add(newrev)

                    continue

                parents = [p for p in parentrevs(newrev) if p != nullrev]

                samebranch = set()