ballistica/tools/bacommon/cloud.py

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

from __future__ import annotations

import datetime
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.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: Annotated[str, IOAttrs('u')]

    # URL to use for overlay-web-browser sign ins.
    url_overlay: Annotated[str, IOAttrs('uo')]

    # Proxy-Login id for querying results.
    proxyid: Annotated[str, IOAttrs('p')]

    # Proxy-Login key for querying results.
    proxykey: Annotated[str, IOAttrs('k')]


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

    proxyid: Annotated[str, IOAttrs('p')]
    proxykey: Annotated[str, IOAttrs('k')]

    @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: Annotated[State, IOAttrs('s')]

    # On success, these will be filled out.
    credentials: Annotated[str | None, IOAttrs('tk')]


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

    proxyid: Annotated[str, IOAttrs('p')]


@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: Annotated[int, IOAttrs('f')]

    @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: Annotated[int, IOAttrs('f')]


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

    description: Annotated[str, IOAttrs('c')]

    @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: Annotated[bool, IOAttrs('v')]
    message: Annotated[str | None, IOAttrs('m', store_default=False)] = None
    legacy_code: Annotated[str | None, IOAttrs('l', store_default=False)] = None


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

    manifest: Annotated[DirectoryManifest, IOAttrs('m')]
    iteration: Annotated[int, IOAttrs('i')] = 0
    total_deletes: Annotated[int, IOAttrs('tdels')] = 0
    total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
    total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None


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

    workspaceid: Annotated[str, IOAttrs('w')]
    state: Annotated[WorkspaceFetchState, IOAttrs('s')]

    @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: Annotated[WorkspaceFetchState, IOAttrs('s')]
    deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
        default_factory=list
    )
    downloads_inline: Annotated[
        dict[str, bytes], IOAttrs('dinl', store_default=False)
    ] = field(default_factory=dict)

    done: Annotated[bool, IOAttrs('d')] = 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: Annotated[str | None, IOAttrs('u')]


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

    login_type: Annotated[LoginType, IOAttrs('l')]
    sign_in_token: Annotated[str, IOAttrs('t')]

    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
    description: Annotated[str, IOAttrs('d', soft_default='-')]
    apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]

    @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: Annotated[str | None, IOAttrs('c')]


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

    weblocation: Annotated[WebLocation, IOAttrs('l')] = (
        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: Annotated[str | None, IOAttrs('u')]


@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: Annotated[str, IOAttrs('id')]

    # Overall result; all data is undefined if not SUCCESS.
    result: Annotated[Result, IOAttrs('r')]

    tokens: Annotated[int, IOAttrs('t')]
    gold_pass: Annotated[bool, IOAttrs('g')]

    available_purchases: Annotated[list[Purchase], IOAttrs('p')]
    token_info_url: Annotated[str, IOAttrs('tiu')]


@ioprepped
@dataclass
class BSPrivatePartyMessage(Message):
    """Message asking about info we need for private-party UI."""

    need_datacode: Annotated[bool, IOAttrs('d')]

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


@ioprepped
@dataclass
class BSPrivatePartyResponse(Response):
    """Here's that private party UI info you asked for, boss."""

    success: Annotated[bool, IOAttrs('s')]
    tokens: Annotated[int, IOAttrs('t')]
    gold_pass: Annotated[bool, IOAttrs('g')]
    datacode: Annotated[str | None, IOAttrs('d')]


class BSClassicChestAppearance(Enum):
    """Appearances bombsquad classic chests can have."""

    UNKNOWN = 'u'
    DEFAULT = 'd'


@ioprepped
@dataclass
class BSClassicAccountLiveData:
    """Account related data kept up to date live for classic app mode."""

    @dataclass
    class Chest:
        """A lovely chest."""

        appearance: Annotated[
            BSClassicChestAppearance,
            IOAttrs('a', enum_fallback=BSClassicChestAppearance.UNKNOWN),
        ]
        unlock_time: Annotated[datetime.datetime, IOAttrs('t')]
        ad_allow_time: Annotated[datetime.datetime | None, IOAttrs('at')]

    class LeagueType(Enum):
        """Type of league we are in."""

        BRONZE = 'b'
        SILVER = 's'
        GOLD = 'g'
        DIAMOND = 'd'

    tickets: Annotated[int, IOAttrs('ti')]

    tokens: Annotated[int, IOAttrs('to')]
    gold_pass: Annotated[bool, IOAttrs('g')]

    achievements: Annotated[int, IOAttrs('a')]
    achievements_total: Annotated[int, IOAttrs('at')]

    league_type: Annotated[LeagueType | None, IOAttrs('lt')]
    league_num: Annotated[int | None, IOAttrs('ln')]
    league_rank: Annotated[int | None, IOAttrs('lr')]

    level: Annotated[int, IOAttrs('lv')]
    xp: Annotated[int, IOAttrs('xp')]
    xpmax: Annotated[int, IOAttrs('xpm')]

    inbox_count: Annotated[int, IOAttrs('ibc')]
    inbox_count_is_max: Annotated[bool, IOAttrs('ibcm')]

    chests: Annotated[dict[str, Chest], IOAttrs('c')]


class BSInboxEntryType(Enum):
    """Types of entries that can be in an inbox."""

    UNKNOWN = 'u'  # Entry types we don't support will be this.
    SIMPLE = 's'
    CLAIM = 'c'
    CLAIM_DISCARD = 'cd'


@ioprepped
@dataclass
class BSInboxEntry:
    """Single message in an inbox."""

    type: Annotated[
        BSInboxEntryType, IOAttrs('t', enum_fallback=BSInboxEntryType.UNKNOWN)
    ]
    id: Annotated[str, IOAttrs('i')]
    createtime: Annotated[datetime.datetime, IOAttrs('c')]

    # If clients don't support format_version of a message they will
    # display 'app needs to be updated to show this'.
    format_version: Annotated[int, IOAttrs('f', soft_default=1)]

    # These have soft defaults so can be removed in the future if desired.
    message: Annotated[str, IOAttrs('m', soft_default='(invalid message)')]
    subs: Annotated[list[str], IOAttrs('s', soft_default_factory=list)]


@ioprepped
@dataclass
class BSInboxRequestMessage(Message):
    """Message requesting our inbox."""

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


@ioprepped
@dataclass
class BSInboxRequestResponse(Response):
    """Here's that inbox contents you asked for, boss."""

    entries: Annotated[list[BSInboxEntry], IOAttrs('m')]

    # Printable error if something goes wrong.
    error: Annotated[str | None, IOAttrs('e')] = None


@ioprepped
@dataclass
class BSChestInfoMessage(Message):
    """Request info about a chest."""

    chest_id: Annotated[str, IOAttrs('i')]

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


@ioprepped
@dataclass
class BSChestInfoResponse(Response):
    """Here's that inbox contents you asked for, boss."""

    @dataclass
    class Chest:
        """A lovely chest."""

        appearance: Annotated[
            BSClassicChestAppearance,
            IOAttrs('a', enum_fallback=BSClassicChestAppearance.UNKNOWN),
        ]

        # How much to unlock *now*.
        unlock_tokens: Annotated[int, IOAttrs('tk')]

        # When unlocks on its own.
        unlock_time: Annotated[datetime.datetime, IOAttrs('t')]

        # Are ads allowed now?
        ad_allow: Annotated[bool, IOAttrs('aa')]

    chest: Annotated[Chest | None, IOAttrs('c')]


@ioprepped
@dataclass
class BSChestActionMessage(Message):
    """Request action about a chest."""

    class Action(Enum):
        """Types of actions we can request."""

        # Unlocking (for free or with tokens).
        UNLOCK = 'u'

        # Watched an ad to reduce wait.
        AD = 'ad'

    action: Annotated[Action, IOAttrs('a')]

    # Tokens we are paying (only applies to unlock).
    token_payment: Annotated[int, IOAttrs('t')]

    chest_id: Annotated[str, IOAttrs('i')]

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


@ioprepped
@dataclass
class BSChestActionResponse(Response):
    """Here's the results of that action you asked for, boss."""

    # If present, signifies the chest has been opened and we should show
    # the user this stuff that was in it.
    contents: Annotated[list[str] | None, IOAttrs('c')] = None

    # Printable error if something goes wrong.
    error: Annotated[str | None, IOAttrs('e')] = None


class BSInboxEntryProcessType(Enum):
    """Types of processing we can ask for."""

    POSITIVE = 'p'
    NEGATIVE = 'n'


@ioprepped
@dataclass
class BSInboxEntryProcessMessage(Message):
    """Do something to an inbox entry."""

    id: Annotated[str, IOAttrs('i')]
    process_type: Annotated[BSInboxEntryProcessType, IOAttrs('t')]

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


@ioprepped
@dataclass
class BSInboxEntryProcessResponse(Response):
    """Did something to that inbox entry, boss."""

    # Printable error if something goes wrong.
    error: Annotated[str | None, IOAttrs('e')] = None