Source code for bacommon.cloud

# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to cloud functionality."""

from __future__ import annotations

from enum import Enum
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Annotated, override

from efro.message import Message, Response
from efro.dataclassio import ioprepped, IOAttrs
from bacommon.securedata import SecureDataChecker
from bacommon.transfer import DirectoryManifest
from bacommon.login import LoginType

if TYPE_CHECKING:
    pass




class WebLocation(Enum):
    """Set of places we can be directed on ballistica.net."""

    ACCOUNT_EDITOR = 'e'
    ACCOUNT_DELETE_SECTION = 'd'





@ioprepped
@dataclass
class LoginProxyRequestMessage(Message):
    """Request send to the cloud to ask for a login-proxy."""



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [LoginProxyRequestResponse]






@ioprepped
@dataclass
class LoginProxyRequestResponse(Response):
    """Response to a request for a login proxy."""

    # URL to direct the user to for sign in.
    url: str

    # URL to use for overlay-web-browser sign ins.
    url_overlay: str

    # Proxy-Login id for querying results.
    proxyid: str

    # Proxy-Login key for querying results.
    proxykey: str





@ioprepped
@dataclass
class LoginProxyStateQueryMessage(Message):
    """Soo.. how is that login proxy going?"""

    proxyid: str
    proxykey: str



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [LoginProxyStateQueryResponse]






@ioprepped
@dataclass
class LoginProxyStateQueryResponse(Response):
    """Here's the info on that login-proxy you asked about, boss."""



    class State(Enum):
        """States a login-proxy can be in."""

        WAITING = 'waiting'
        SUCCESS = 'success'
        FAIL = 'fail'


    state: State

    # On success, these will be filled out.
    credentials: str | None





@ioprepped
@dataclass
class LoginProxyCompleteMessage(Message):
    """Just so you know, we're done with this proxy."""

    proxyid: str





@ioprepped
@dataclass
class PingMessage(Message):
    """Standard ping."""



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [PingResponse]






@ioprepped
@dataclass
class PingResponse(Response):
    """pong."""





@ioprepped
@dataclass
class TestMessage(Message):
    """Can I get some of that workspace action?"""

    testfoo: int



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [TestResponse]






@ioprepped
@dataclass
class TestResponse(Response):
    """Here's that workspace you asked for, boss."""

    testfoo: int





@ioprepped
@dataclass
class SendInfoMessage(Message):
    """User is using the send-info function"""

    description: str



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [SendInfoResponse]






@ioprepped
@dataclass
class SendInfoResponse(Response):
    """Response to sending into the server."""

    handled: bool
    message: str | None = None
    legacy_code: str | None = None





@ioprepped
@dataclass
class WorkspaceFetchState:
    """Common state data for a workspace fetch."""

    manifest: DirectoryManifest
    iteration: int = 0
    total_deletes: int = 0
    total_downloads: int = 0
    total_up_to_date: int | None = None





@ioprepped
@dataclass
class WorkspaceFetchMessage(Message):
    """Can I get some of that workspace action?"""

    workspaceid: str
    state: WorkspaceFetchState



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [WorkspaceFetchResponse]






@ioprepped
@dataclass
class WorkspaceFetchResponse(Response):
    """Here's that workspace you asked for, boss."""

    state: WorkspaceFetchState
    deletes: list[str] = field(
        default_factory=list
    )
    downloads_inline: dict[str, bytes] = field(default_factory=dict)

    done: bool = False





@ioprepped
@dataclass
class MerchAvailabilityMessage(Message):
    """Can we show merch link?"""



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [MerchAvailabilityResponse]






@ioprepped
@dataclass
class MerchAvailabilityResponse(Response):
    """About that merch..."""

    url: str | None





@ioprepped
@dataclass
class SignInMessage(Message):
    """Can I sign in please?"""

    login_type: LoginType
    sign_in_token: str

    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
    description: str
    apptime: float



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [SignInResponse]






@ioprepped
@dataclass
class SignInResponse(Response):
    """Here's that sign-in result you asked for, boss."""

    credentials: str | None





@ioprepped
@dataclass
class ManageAccountMessage(Message):
    """Message asking for a manage-account url."""

    weblocation: WebLocation = (
        WebLocation.ACCOUNT_EDITOR
    )



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [ManageAccountResponse]






@ioprepped
@dataclass
class ManageAccountResponse(Response):
    """Here's that sign-in result you asked for, boss."""

    url: str | None





@ioprepped
@dataclass
class StoreQueryMessage(Message):
    """Message asking about purchasable stuff and store related state."""



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [StoreQueryResponse]






@ioprepped
@dataclass
class StoreQueryResponse(Response):
    """Here's that store info you asked for, boss."""



    class Result(Enum):
        """Our overall result."""

        SUCCESS = 's'
        ERROR = 'e'




    @dataclass
    class Purchase:
        """Info about a purchasable thing."""

        purchaseid: str


    # Overall result; all data is undefined if not SUCCESS.
    result: Result

    tokens: int
    gold_pass: bool

    available_purchases: list[Purchase]
    token_info_url: str





@ioprepped
@dataclass
class SecureDataCheckMessage(Message):
    """Was this data signed by the master-server?."""

    data: bytes
    signature: bytes



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [SecureDataCheckResponse]






@ioprepped
@dataclass
class SecureDataCheckResponse(Response):
    """Here's the result of that data check, boss."""

    # Whether the data signature was valid.
    result: bool





@ioprepped
@dataclass
class SecureDataCheckerRequest(Message):
    """Can I get a checker over here?."""



    @override
    @classmethod
    def get_response_types(cls) -> list[type[Response] | None]:
        return [SecureDataCheckerResponse]






@ioprepped
@dataclass
class SecureDataCheckerResponse(Response):
    """Here's that checker ya asked for, boss."""

    checker: SecureDataChecker



# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
  Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future