bacommon.cloud

  1# Released under the MIT License. See LICENSE for details.
  2#
  3"""Functionality related to cloud functionality."""
  4
  5from __future__ import annotations
  6from dataclasses import dataclass, field
  7from typing import TYPE_CHECKING, Annotated, override
  8from enum import Enum
  9
 10from efro.message import Message, Response
 11from efro.dataclassio import ioprepped, IOAttrs
 12from bacommon.transfer import DirectoryManifest
 13from bacommon.login import LoginType
 14
 15if TYPE_CHECKING:
 16    pass
 17
 18
 19class WebLocation(Enum):
 20    """Set of places we can be directed on ballistica.net."""
 21
 22    ACCOUNT_EDITOR = 'e'
 23    ACCOUNT_DELETE_SECTION = 'd'
 24
 25
 26@ioprepped
 27@dataclass
 28class LoginProxyRequestMessage(Message):
 29    """Request send to the cloud to ask for a login-proxy."""
 30
 31    @override
 32    @classmethod
 33    def get_response_types(cls) -> list[type[Response] | None]:
 34        return [LoginProxyRequestResponse]
 35
 36
 37@ioprepped
 38@dataclass
 39class LoginProxyRequestResponse(Response):
 40    """Response to a request for a login proxy."""
 41
 42    # URL to direct the user to for sign in.
 43    url: Annotated[str, IOAttrs('u')]
 44
 45    # URL to use for overlay-web-browser sign ins.
 46    url_overlay: Annotated[str, IOAttrs('uo')]
 47
 48    # Proxy-Login id for querying results.
 49    proxyid: Annotated[str, IOAttrs('p')]
 50
 51    # Proxy-Login key for querying results.
 52    proxykey: Annotated[str, IOAttrs('k')]
 53
 54
 55@ioprepped
 56@dataclass
 57class LoginProxyStateQueryMessage(Message):
 58    """Soo.. how is that login proxy going?"""
 59
 60    proxyid: Annotated[str, IOAttrs('p')]
 61    proxykey: Annotated[str, IOAttrs('k')]
 62
 63    @override
 64    @classmethod
 65    def get_response_types(cls) -> list[type[Response] | None]:
 66        return [LoginProxyStateQueryResponse]
 67
 68
 69@ioprepped
 70@dataclass
 71class LoginProxyStateQueryResponse(Response):
 72    """Here's the info on that login-proxy you asked about, boss."""
 73
 74    class State(Enum):
 75        """States a login-proxy can be in."""
 76
 77        WAITING = 'waiting'
 78        SUCCESS = 'success'
 79        FAIL = 'fail'
 80
 81    state: Annotated[State, IOAttrs('s')]
 82
 83    # On success, these will be filled out.
 84    credentials: Annotated[str | None, IOAttrs('tk')]
 85
 86
 87@ioprepped
 88@dataclass
 89class LoginProxyCompleteMessage(Message):
 90    """Just so you know, we're done with this proxy."""
 91
 92    proxyid: Annotated[str, IOAttrs('p')]
 93
 94
 95@ioprepped
 96@dataclass
 97class PingMessage(Message):
 98    """Standard ping."""
 99
100    @override
101    @classmethod
102    def get_response_types(cls) -> list[type[Response] | None]:
103        return [PingResponse]
104
105
106@ioprepped
107@dataclass
108class PingResponse(Response):
109    """pong."""
110
111
112@ioprepped
113@dataclass
114class TestMessage(Message):
115    """Can I get some of that workspace action?"""
116
117    testfoo: Annotated[int, IOAttrs('f')]
118
119    @override
120    @classmethod
121    def get_response_types(cls) -> list[type[Response] | None]:
122        return [TestResponse]
123
124
125@ioprepped
126@dataclass
127class TestResponse(Response):
128    """Here's that workspace you asked for, boss."""
129
130    testfoo: Annotated[int, IOAttrs('f')]
131
132
133@ioprepped
134@dataclass
135class SendInfoMessage(Message):
136    """User is using the send-info function"""
137
138    description: Annotated[str, IOAttrs('c')]
139
140    @override
141    @classmethod
142    def get_response_types(cls) -> list[type[Response] | None]:
143        return [SendInfoResponse]
144
145
146@ioprepped
147@dataclass
148class SendInfoResponse(Response):
149    """Response to sending into the server."""
150
151    handled: Annotated[bool, IOAttrs('v')]
152    message: Annotated[str | None, IOAttrs('m', store_default=False)] = None
153    legacy_code: Annotated[str | None, IOAttrs('l', store_default=False)] = None
154
155
156@ioprepped
157@dataclass
158class WorkspaceFetchState:
159    """Common state data for a workspace fetch."""
160
161    manifest: Annotated[DirectoryManifest, IOAttrs('m')]
162    iteration: Annotated[int, IOAttrs('i')] = 0
163    total_deletes: Annotated[int, IOAttrs('tdels')] = 0
164    total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
165    total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None
166
167
168@ioprepped
169@dataclass
170class WorkspaceFetchMessage(Message):
171    """Can I get some of that workspace action?"""
172
173    workspaceid: Annotated[str, IOAttrs('w')]
174    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
175
176    @override
177    @classmethod
178    def get_response_types(cls) -> list[type[Response] | None]:
179        return [WorkspaceFetchResponse]
180
181
182@ioprepped
183@dataclass
184class WorkspaceFetchResponse(Response):
185    """Here's that workspace you asked for, boss."""
186
187    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
188    deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
189        default_factory=list
190    )
191    downloads_inline: Annotated[
192        dict[str, bytes], IOAttrs('dinl', store_default=False)
193    ] = field(default_factory=dict)
194
195    done: Annotated[bool, IOAttrs('d')] = False
196
197
198@ioprepped
199@dataclass
200class MerchAvailabilityMessage(Message):
201    """Can we show merch link?"""
202
203    @override
204    @classmethod
205    def get_response_types(cls) -> list[type[Response] | None]:
206        return [MerchAvailabilityResponse]
207
208
209@ioprepped
210@dataclass
211class MerchAvailabilityResponse(Response):
212    """About that merch..."""
213
214    url: Annotated[str | None, IOAttrs('u')]
215
216
217@ioprepped
218@dataclass
219class SignInMessage(Message):
220    """Can I sign in please?"""
221
222    login_type: Annotated[LoginType, IOAttrs('l')]
223    sign_in_token: Annotated[str, IOAttrs('t')]
224
225    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
226    description: Annotated[str, IOAttrs('d', soft_default='-')]
227    apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]
228
229    @override
230    @classmethod
231    def get_response_types(cls) -> list[type[Response] | None]:
232        return [SignInResponse]
233
234
235@ioprepped
236@dataclass
237class SignInResponse(Response):
238    """Here's that sign-in result you asked for, boss."""
239
240    credentials: Annotated[str | None, IOAttrs('c')]
241
242
243@ioprepped
244@dataclass
245class ManageAccountMessage(Message):
246    """Message asking for a manage-account url."""
247
248    weblocation: Annotated[WebLocation, IOAttrs('l')] = (
249        WebLocation.ACCOUNT_EDITOR
250    )
251
252    @override
253    @classmethod
254    def get_response_types(cls) -> list[type[Response] | None]:
255        return [ManageAccountResponse]
256
257
258@ioprepped
259@dataclass
260class ManageAccountResponse(Response):
261    """Here's that sign-in result you asked for, boss."""
262
263    url: Annotated[str | None, IOAttrs('u')]
264
265
266@ioprepped
267@dataclass
268class StoreQueryMessage(Message):
269    """Message asking about purchasable stuff and store related state."""
270
271    @override
272    @classmethod
273    def get_response_types(cls) -> list[type[Response] | None]:
274        return [StoreQueryResponse]
275
276
277@ioprepped
278@dataclass
279class StoreQueryResponse(Response):
280    """Here's that store info you asked for, boss."""
281
282    class Result(Enum):
283        """Our overall result."""
284
285        SUCCESS = 's'
286        ERROR = 'e'
287
288    @dataclass
289    class Purchase:
290        """Info about a purchasable thing."""
291
292        purchaseid: Annotated[str, IOAttrs('id')]
293
294    # Overall result; all data is undefined if not SUCCESS.
295    result: Annotated[Result, IOAttrs('r')]
296
297    tokens: Annotated[int, IOAttrs('t')]
298    gold_pass: Annotated[bool, IOAttrs('g')]
299
300    available_purchases: Annotated[list[Purchase], IOAttrs('p')]
301    token_info_url: Annotated[str, IOAttrs('tiu')]
302
303
304@ioprepped
305@dataclass
306class BSPrivatePartyMessage(Message):
307    """Message asking about info we need for private-party UI."""
308
309    need_datacode: Annotated[bool, IOAttrs('d')]
310
311    @override
312    @classmethod
313    def get_response_types(cls) -> list[type[Response] | None]:
314        return [BSPrivatePartyResponse]
315
316
317@ioprepped
318@dataclass
319class BSPrivatePartyResponse(Response):
320    """Here's that private party UI info you asked for, boss."""
321
322    success: Annotated[bool, IOAttrs('s')]
323    tokens: Annotated[int, IOAttrs('t')]
324    gold_pass: Annotated[bool, IOAttrs('g')]
325    datacode: Annotated[str | None, IOAttrs('d')]