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
from pathlib import Path
from enum import Enum