# repository.py - Interfaces and base classes for repositories and peers.

#

# Copyright 2017 Gregory Szorc <gregory.szorc@gmail.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 __future__ import annotations

import abc

import typing

from typing import (

    Any,

    Callable,

    Collection,

    Iterable,

    Iterator,

    Mapping,

    Optional,

    Protocol,

    Set,

)

from ..i18n import _

from .. import error

if typing.TYPE_CHECKING:

    from typing import (

        ByteString,  # TODO: change to Buffer for 3.14

    )

    from . import (

        dirstate as intdirstate,

        matcher,

        misc,

    )

    # TODO: make a protocol class for this

    NodeConstants = Any

# Local repository feature string.

# Revlogs are being used for file storage.

REPO_FEATURE_REVLOG_FILE_STORAGE = b'revlogfilestorage'

# The storage part of the repository is shared from an external source.

REPO_FEATURE_SHARED_STORAGE = b'sharedstore'

# LFS supported for backing file storage.

REPO_FEATURE_LFS = b'lfs'

# Repository supports being stream cloned.

REPO_FEATURE_STREAM_CLONE = b'streamclone'

# Repository supports (at least) some sidedata to be stored

REPO_FEATURE_SIDE_DATA = b'side-data'

# Files storage may lack data for all ancestors.

REPO_FEATURE_SHALLOW_FILE_STORAGE = b'shallowfilestorage'

REVISION_FLAG_CENSORED = 1 << 15

REVISION_FLAG_ELLIPSIS = 1 << 14

REVISION_FLAG_EXTSTORED = 1 << 13

REVISION_FLAG_HASCOPIESINFO = 1 << 12

REVISION_FLAGS_KNOWN = (

    REVISION_FLAG_CENSORED

    | REVISION_FLAG_ELLIPSIS

    | REVISION_FLAG_EXTSTORED

    | REVISION_FLAG_HASCOPIESINFO

)

CG_DELTAMODE_STD = b'default'

CG_DELTAMODE_PREV = b'previous'

CG_DELTAMODE_FULL = b'fulltext'

CG_DELTAMODE_P1 = b'p1'

## Cache related constants:

#

# Used to control which cache should be warmed in a repo.updatecaches(…) call.

# Warm branchmaps of all known repoview's filter-level

CACHE_BRANCHMAP_ALL = b"branchmap-all"

# Warm branchmaps of repoview's filter-level used by server

CACHE_BRANCHMAP_SERVED = b"branchmap-served"

# Warm internal changelog cache (eg: persistent nodemap)

CACHE_CHANGELOG_CACHE = b"changelog-cache"

# check of a branchmap can use the "pure topo" mode

CACHE_BRANCHMAP_DETECT_PURE_TOPO = b"branchmap-detect-pure-topo"

# Warm full manifest cache

CACHE_FULL_MANIFEST = b"full-manifest"

# Warm file-node-tags cache

CACHE_FILE_NODE_TAGS = b"file-node-tags"

# Warm internal manifestlog cache (eg: persistent nodemap)

CACHE_MANIFESTLOG_CACHE = b"manifestlog-cache"

# Warn rev branch cache

CACHE_REV_BRANCH = b"rev-branch-cache"

# Warm tags' cache for default repoview'

CACHE_TAGS_DEFAULT = b"tags-default"

# Warm tags' cache for  repoview's filter-level used by server

CACHE_TAGS_SERVED = b"tags-served"

# the cache to warm by default after a simple transaction

# (this is a mutable set to let extension update it)

CACHES_DEFAULT = {

    CACHE_BRANCHMAP_SERVED,

}

# the caches to warm when warming all of them

# (this is a mutable set to let extension update it)

CACHES_ALL = {

    CACHE_BRANCHMAP_SERVED,

    CACHE_BRANCHMAP_ALL,

    CACHE_BRANCHMAP_DETECT_PURE_TOPO,

    CACHE_REV_BRANCH,

    CACHE_CHANGELOG_CACHE,

    CACHE_FILE_NODE_TAGS,

    CACHE_FULL_MANIFEST,

    CACHE_MANIFESTLOG_CACHE,

    CACHE_TAGS_DEFAULT,

    CACHE_TAGS_SERVED,

}

# the cache to warm by default on simple call

# (this is a mutable set to let extension update it)

CACHES_POST_CLONE = CACHES_ALL.copy()

CACHES_POST_CLONE.discard(CACHE_FILE_NODE_TAGS)

CACHES_POST_CLONE.discard(CACHE_REV_BRANCH)

class _ipeerconnection(Protocol):

    """Represents a "connection" to a repository.

    This is the base interface for representing a connection to a repository.

    It holds basic properties and methods applicable to all peer types.

    This is not a complete interface definition and should not be used

    outside of this module.

    """

    ui: Ui

    """ui.ui instance"""

    path: Optional[misc.IPath]

    """a urlutil.path instance or None"""

    @abc.abstractmethod

    def url(self):

        """Returns a URL string representing this peer.

        Currently, implementations expose the raw URL used to construct the

        instance. It may contain credentials as part of the URL. The

        expectations of the value aren't well-defined and this could lead to

        data leakage.

        TODO audit/clean consumers and more clearly define the contents of this

        value.

        """

    @abc.abstractmethod

    def local(self):

        """Returns a local repository instance.

        If the peer represents a local repository, returns an object that

        can be used to interface with it. Otherwise returns ``None``.

        """

    @abc.abstractmethod

    def canpush(self):

        """Returns a boolean indicating if this peer can be pushed to."""

    @abc.abstractmethod

    def close(self):

        """Close the connection to this peer.

        This is called when the peer will no longer be used. Resources

        associated with the peer should be cleaned up.

        """

class ipeercapabilities(Protocol):

    """Peer sub-interface related to capabilities."""

    @abc.abstractmethod

    def capable(self, name: bytes) -> bool | bytes:

        """Determine support for a named capability.

        Returns ``False`` if capability not supported.

        Returns ``True`` if boolean capability is supported. Returns a string

        if capability support is non-boolean.

        Capability strings may or may not map to wire protocol capabilities.

        """

    @abc.abstractmethod

    def capabilities(self) -> Set[bytes]:

        """Obtain capabilities of the peer.

        Returns a set of string capabilities.

        """

    @abc.abstractmethod

    def requirecap(self, name: bytes, purpose: bytes) -> None:

        """Require a capability to be present.

        Raises a ``CapabilityError`` if the capability isn't present.

        """

class ipeercommands(Protocol):

    """Client-side interface for communicating over the wire protocol.

    This interface is used as a gateway to the Mercurial wire protocol.

    methods commonly call wire protocol commands of the same name.

    """

    @abc.abstractmethod

    def branchmap(self):

        """Obtain heads in named branches.

        Returns a dict mapping branch name to an iterable of nodes that are

        heads on that branch.

        """

    @abc.abstractmethod

    def get_cached_bundle_inline(self, path):

        """Retrieve a clonebundle across the wire.

        Returns a chunkbuffer

        """

    @abc.abstractmethod

    def clonebundles(self):

        """Obtains the clone bundles manifest for the repo.

        Returns the manifest as unparsed bytes.

        """

    @abc.abstractmethod

    def debugwireargs(self, one, two, three=None, four=None, five=None):

        """Used to facilitate debugging of arguments passed over the wire."""

    @abc.abstractmethod

    def getbundle(self, source, **kwargs):

        """Obtain remote repository data as a bundle.

        This command is how the bulk of repository data is transferred from

        the peer to the local repository

        Returns a generator of bundle data.

        """

    @abc.abstractmethod

    def heads(self):

        """Determine all known head revisions in the peer.

        Returns an iterable of binary nodes.

        """

    @abc.abstractmethod

    def known(self, nodes):

        """Determine whether multiple nodes are known.

        Accepts an iterable of nodes whose presence to check for.

        Returns an iterable of booleans indicating of the corresponding node

        at that index is known to the peer.

        """

    @abc.abstractmethod

    def listkeys(self, namespace):

        """Obtain all keys in a pushkey namespace.

        Returns an iterable of key names.

        """

    @abc.abstractmethod

    def lookup(self, key):

        """Resolve a value to a known revision.

        Returns a binary node of the resolved revision on success.

        """

    @abc.abstractmethod

    def pushkey(self, namespace, key, old, new):

        """Set a value using the ``pushkey`` protocol.

        Arguments correspond to the pushkey namespace and key to operate on and

        the old and new values for that key.

        Returns a string with the peer result. The value inside varies by the

        namespace.

        """

    @abc.abstractmethod

    def stream_out(self):

        """Obtain streaming clone data.

        Successful result should be a generator of data chunks.

        """

    @abc.abstractmethod

    def unbundle(self, bundle, heads, url):

        """Transfer repository data to the peer.

        This is how the bulk of data during a push is transferred.

        Returns the integer number of heads added to the peer.

        """

class ipeerlegacycommands(Protocol):

    """Interface for implementing support for legacy wire protocol commands.

    Wire protocol commands transition to legacy status when they are no longer

    used by modern clients. To facilitate identifying which commands are

    legacy, the interfaces are split.

    """

    @abc.abstractmethod

    def between(self, pairs):

        """Obtain nodes between pairs of nodes.

        ``pairs`` is an iterable of node pairs.

        Returns an iterable of iterables of nodes corresponding to each

        requested pair.

        """

    @abc.abstractmethod

    def branches(self, nodes):

        """Obtain ancestor changesets of specific nodes back to a branch point.

        For each requested node, the peer finds the first ancestor node that is

        a DAG root or is a merge.

        Returns an iterable of iterables with the resolved values for each node.

        """

    @abc.abstractmethod

    def changegroup(self, nodes, source):

        """Obtain a changegroup with data for descendants of specified nodes."""

    @abc.abstractmethod

    def changegroupsubset(self, bases, heads, source):

        pass

class ipeercommandexecutor(Protocol):

    """Represents a mechanism to execute remote commands.

    This is the primary interface for requesting that wire protocol commands

    be executed. Instances of this interface are active in a context manager

    and have a well-defined lifetime. When the context manager exits, all

    outstanding requests are waited on.

    """

    @abc.abstractmethod

    def callcommand(self, name, args):

        """Request that a named command be executed.

        Receives the command name and a dictionary of command arguments.

        Returns a ``concurrent.futures.Future`` that will resolve to the

        result of that command request. That exact value is left up to

        the implementation and possibly varies by command.

        Not all commands can coexist with other commands in an executor

        instance: it depends on the underlying wire protocol transport being

        used and the command itself.

        Implementations MAY call ``sendcommands()`` automatically if the

        requested command can not coexist with other commands in this executor.

        Implementations MAY call ``sendcommands()`` automatically when the

        future's ``result()`` is called. So, consumers using multiple

        commands with an executor MUST ensure that ``result()`` is not called

        until all command requests have been issued.

        """

    @abc.abstractmethod

    def sendcommands(self):

        """Trigger submission of queued command requests.

        Not all transports submit commands as soon as they are requested to

        run. When called, this method forces queued command requests to be

        issued. It will no-op if all commands have already been sent.

        When called, no more new commands may be issued with this executor.

        """

    @abc.abstractmethod

    def close(self):

        """Signal that this command request is finished.

        When called, no more new commands may be issued. All outstanding

        commands that have previously been issued are waited on before

        returning. This not only includes waiting for the futures to resolve,

        but also waiting for all response data to arrive. In other words,

        calling this waits for all on-wire state for issued command requests

        to finish.

        When used as a context manager, this method is called when exiting the

        context manager.

        This method may call ``sendcommands()`` if there are buffered commands.

        """

class ipeerrequests(Protocol):

    """Interface for executing commands on a peer."""

    limitedarguments: bool

    """True if the peer cannot receive large argument value for commands."""

    @abc.abstractmethod

    def commandexecutor(self):

        """A context manager that resolves to an ipeercommandexecutor.

        The object this resolves to can be used to issue command requests

        to the peer.

        Callers should call its ``callcommand`` method to issue command

        requests.

        A new executor should be obtained for each distinct set of commands

        (possibly just a single command) that the consumer wants to execute

        as part of a single operation or round trip. This is because some

        peers are half-duplex and/or don't support persistent connections.

        e.g. in the case of HTTP peers, commands sent to an executor represent

        a single HTTP request. While some peers may support multiple command

        sends over the wire per executor, consumers need to code to the least

        capable peer. So it should be assumed that command executors buffer

        called commands until they are told to send them and that each

        command executor could result in a new connection or wire-level request

        being issued.

        """

# TODO: make this a Protocol class when 3.11 is the minimum supported version?

class peer(_ipeerconnection, ipeercapabilities, ipeerrequests):

    """Unified interface for peer repositories.

    All peer instances must conform to this interface.

    """

    limitedarguments: bool = False

    path: misc.IPath | None

    ui: Ui

    def __init__(

        self,

        ui: Ui,

        path: misc.IPath | None = None,

        remotehidden: bool = False,

    ) -> None:

        self.ui = ui

        self.path = path

    def capable(self, name: bytes) -> bool | bytes:

        caps = self.capabilities()

        if name in caps:

            return True

        name = b'%s=' % name

        for cap in caps:

            if cap.startswith(name):

                return cap[len(name) :]

        return False

    def requirecap(self, name: bytes, purpose: bytes) -> None:

        if self.capable(name):

            return

        raise error.CapabilityError(

            _(

                b'cannot %s; remote repository does not support the '

                b'\'%s\' capability'

            )

            % (purpose, name)

        )