Source code for bacommon.bs._msg

# Released under the MIT License. See LICENSE for details.
#
"""BombSquad specific bits."""

from __future__ import annotations

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

from efro.dataclassio import ioprepped, IOAttrs
from efro.message import Message, Response

from bacommon.bs._displayitem import DisplayItemWrapper
from bacommon.bs._clienteffect import ClientEffect
from bacommon.bs._clouddialog import CloudDialogAction, CloudDialogWrapper
from bacommon.bs._chest import ClassicChestAppearance




@ioprepped
@dataclass
class ChestActionMessage(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: Action

    # Tokens we are paying (only applies to unlock).
    token_payment: int

    chest_id: str



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






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

    # Tokens that were actually charged.
    tokens_charged: int = 0

    # If present, signifies the chest has been opened and we should show
    # the user this stuff that was in it.
    contents: list[DisplayItemWrapper] | None = None

    # If contents are present, which of the chest's prize-sets they
    # represent.
    prizeindex: int = 0

    # Printable error if something goes wrong.
    error: str | None = None

    # Printable warning. Shown in orange with an error sound. Does not
    # mean the action failed; only that there's something to tell the
    # users such as 'It looks like you are faking ad views; stop it or
    # you won't have ad options anymore.'
    warning: str | None = None

    # Printable success message. Shown in green with a cash-register
    # sound. Can be used for things like successful wait reductions via
    # ad views. Used in builds earlier than 22311; can remove once
    # 22311+ is ubiquitous.
    success_msg: str | None = None

    # Effects to show on the client. Replaces warning and success_msg in
    # build 22311 or newer.
    effects: list[ClientEffect] = field(default_factory=list)





@ioprepped
@dataclass
class CloudDialogActionMessage(Message):
    """Do something to a client ui."""

    id: str
    action: CloudDialogAction



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






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



    class ErrorType(Enum):
        """Types of errors that may have occurred."""

        # Probably a future error type we don't recognize.
        UNKNOWN = 'u'

        # Something went wrong on the server, but specifics are not
        # relevant.
        INTERNAL = 'i'

        # The entry expired on the server. In various cases such as 'ok'
        # buttons this can generally be ignored.
        EXPIRED = 'e'


    error_type: ErrorType | None

    # User facing error message in the case of errors.
    error_message: str | None

    effects: list[ClientEffect]





@ioprepped
@dataclass
class GetClassicPurchasesMessage(Message):
    """Asking for current account's classic purchases."""



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






@ioprepped
@dataclass
class GetClassicPurchasesResponse(Response):
    """Here's those classic purchases ya asked for boss."""

    purchases: set[str]





@ioprepped
@dataclass
class GlobalProfileCheckMessage(Message):
    """Is this global profile name available?"""

    name: str



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






@ioprepped
@dataclass
class GlobalProfileCheckResponse(Response):
    """Here's that profile check ya asked for boss."""

    available: bool
    ticket_cost: int





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



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






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

    wrappers: list[CloudDialogWrapper]

    # Printable error if something goes wrong.
    error: str | None = None





@ioprepped
@dataclass
class LegacyRequest(Message):
    """A generic request for the legacy master server."""

    request: str
    request_type: str
    user_agent_string: str
    data: str



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






@ioprepped
@dataclass
class LegacyResponse(Response):
    """Response for generic legacy request."""

    data: str | None
    zipped: bool





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

    chest_id: str



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






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



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



        @dataclass
        class PrizeSet:
            """A possible set of prizes for this chest."""

            weight: float
            contents: list[DisplayItemWrapper]


        appearance: ClassicChestAppearance

        # How much it costs to unlock *now*.
        unlock_tokens: int

        # When it unlocks on its own.
        unlock_time: datetime.datetime

        # Possible prizes we contain.
        prizesets: list[PrizeSet]

        # Are ads allowed now?
        ad_allow: bool


    chest: Chest | None
    user_tokens: int | None





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

    need_datacode: bool



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






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

    success: bool
    tokens: int
    gold_pass: bool
    datacode: str | None





@ioprepped
@dataclass
class ScoreSubmitMessage(Message):
    """Let the server know we got some score in something."""

    score_token: str



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






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

    # Things we should show on our end.
    effects: list[ClientEffect]





@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 info to the server."""

    handled: bool
    message: str | None = None
    effects: list[ClientEffect] = field(default_factory=list)
    legacy_code: str | None = None



# 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