bacommon.cloud

  1# Released under the MIT License. See LICENSE for details.
  2#
  3"""Functionality related to cloud functionality."""
  4
  5from __future__ import annotations
  6
  7import datetime
  8from enum import Enum
  9from dataclasses import dataclass, field
 10from typing import TYPE_CHECKING, Annotated, override
 11
 12from efro.message import Message, Response
 13from efro.dataclassio import ioprepped, IOAttrs
 14from bacommon.transfer import DirectoryManifest
 15from bacommon.login import LoginType
 16
 17if TYPE_CHECKING:
 18    pass
 19
 20
 21class WebLocation(Enum):
 22    """Set of places we can be directed on ballistica.net."""
 23
 24    ACCOUNT_EDITOR = 'e'
 25    ACCOUNT_DELETE_SECTION = 'd'
 26
 27
 28@ioprepped
 29@dataclass
 30class LoginProxyRequestMessage(Message):
 31    """Request send to the cloud to ask for a login-proxy."""
 32
 33    @override
 34    @classmethod
 35    def get_response_types(cls) -> list[type[Response] | None]:
 36        return [LoginProxyRequestResponse]
 37
 38
 39@ioprepped
 40@dataclass
 41class LoginProxyRequestResponse(Response):
 42    """Response to a request for a login proxy."""
 43
 44    # URL to direct the user to for sign in.
 45    url: Annotated[str, IOAttrs('u')]
 46
 47    # URL to use for overlay-web-browser sign ins.
 48    url_overlay: Annotated[str, IOAttrs('uo')]
 49
 50    # Proxy-Login id for querying results.
 51    proxyid: Annotated[str, IOAttrs('p')]
 52
 53    # Proxy-Login key for querying results.
 54    proxykey: Annotated[str, IOAttrs('k')]
 55
 56
 57@ioprepped
 58@dataclass
 59class LoginProxyStateQueryMessage(Message):
 60    """Soo.. how is that login proxy going?"""
 61
 62    proxyid: Annotated[str, IOAttrs('p')]
 63    proxykey: Annotated[str, IOAttrs('k')]
 64
 65    @override
 66    @classmethod
 67    def get_response_types(cls) -> list[type[Response] | None]:
 68        return [LoginProxyStateQueryResponse]
 69
 70
 71@ioprepped
 72@dataclass
 73class LoginProxyStateQueryResponse(Response):
 74    """Here's the info on that login-proxy you asked about, boss."""
 75
 76    class State(Enum):
 77        """States a login-proxy can be in."""
 78
 79        WAITING = 'waiting'
 80        SUCCESS = 'success'
 81        FAIL = 'fail'
 82
 83    state: Annotated[State, IOAttrs('s')]
 84
 85    # On success, these will be filled out.
 86    credentials: Annotated[str | None, IOAttrs('tk')]
 87
 88
 89@ioprepped
 90@dataclass
 91class LoginProxyCompleteMessage(Message):
 92    """Just so you know, we're done with this proxy."""
 93
 94    proxyid: Annotated[str, IOAttrs('p')]
 95
 96
 97@ioprepped
 98@dataclass
 99class PingMessage(Message):
100    """Standard ping."""
101
102    @override
103    @classmethod
104    def get_response_types(cls) -> list[type[Response] | None]:
105        return [PingResponse]
106
107
108@ioprepped
109@dataclass
110class PingResponse(Response):
111    """pong."""
112
113
114@ioprepped
115@dataclass
116class TestMessage(Message):
117    """Can I get some of that workspace action?"""
118
119    testfoo: Annotated[int, IOAttrs('f')]
120
121    @override
122    @classmethod
123    def get_response_types(cls) -> list[type[Response] | None]:
124        return [TestResponse]
125
126
127@ioprepped
128@dataclass
129class TestResponse(Response):
130    """Here's that workspace you asked for, boss."""
131
132    testfoo: Annotated[int, IOAttrs('f')]
133
134
135@ioprepped
136@dataclass
137class SendInfoMessage(Message):
138    """User is using the send-info function"""
139
140    description: Annotated[str, IOAttrs('c')]
141
142    @override
143    @classmethod
144    def get_response_types(cls) -> list[type[Response] | None]:
145        return [SendInfoResponse]
146
147
148@ioprepped
149@dataclass
150class SendInfoResponse(Response):
151    """Response to sending into the server."""
152
153    handled: Annotated[bool, IOAttrs('v')]
154    message: Annotated[str | None, IOAttrs('m', store_default=False)] = None
155    legacy_code: Annotated[str | None, IOAttrs('l', store_default=False)] = None
156
157
158@ioprepped
159@dataclass
160class WorkspaceFetchState:
161    """Common state data for a workspace fetch."""
162
163    manifest: Annotated[DirectoryManifest, IOAttrs('m')]
164    iteration: Annotated[int, IOAttrs('i')] = 0
165    total_deletes: Annotated[int, IOAttrs('tdels')] = 0
166    total_downloads: Annotated[int, IOAttrs('tdlds')] = 0
167    total_up_to_date: Annotated[int | None, IOAttrs('tunmd')] = None
168
169
170@ioprepped
171@dataclass
172class WorkspaceFetchMessage(Message):
173    """Can I get some of that workspace action?"""
174
175    workspaceid: Annotated[str, IOAttrs('w')]
176    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
177
178    @override
179    @classmethod
180    def get_response_types(cls) -> list[type[Response] | None]:
181        return [WorkspaceFetchResponse]
182
183
184@ioprepped
185@dataclass
186class WorkspaceFetchResponse(Response):
187    """Here's that workspace you asked for, boss."""
188
189    state: Annotated[WorkspaceFetchState, IOAttrs('s')]
190    deletes: Annotated[list[str], IOAttrs('dlt', store_default=False)] = field(
191        default_factory=list
192    )
193    downloads_inline: Annotated[
194        dict[str, bytes], IOAttrs('dinl', store_default=False)
195    ] = field(default_factory=dict)
196
197    done: Annotated[bool, IOAttrs('d')] = False
198
199
200@ioprepped
201@dataclass
202class MerchAvailabilityMessage(Message):
203    """Can we show merch link?"""
204
205    @override
206    @classmethod
207    def get_response_types(cls) -> list[type[Response] | None]:
208        return [MerchAvailabilityResponse]
209
210
211@ioprepped
212@dataclass
213class MerchAvailabilityResponse(Response):
214    """About that merch..."""
215
216    url: Annotated[str | None, IOAttrs('u')]
217
218
219@ioprepped
220@dataclass
221class SignInMessage(Message):
222    """Can I sign in please?"""
223
224    login_type: Annotated[LoginType, IOAttrs('l')]
225    sign_in_token: Annotated[str, IOAttrs('t')]
226
227    # For debugging. Can remove soft_default once build 20988+ is ubiquitous.
228    description: Annotated[str, IOAttrs('d', soft_default='-')]
229    apptime: Annotated[float, IOAttrs('at', soft_default=-1.0)]
230
231    @override
232    @classmethod
233    def get_response_types(cls) -> list[type[Response] | None]:
234        return [SignInResponse]
235
236
237@ioprepped
238@dataclass
239class SignInResponse(Response):
240    """Here's that sign-in result you asked for, boss."""
241
242    credentials: Annotated[str | None, IOAttrs('c')]
243
244
245@ioprepped
246@dataclass
247class ManageAccountMessage(Message):
248    """Message asking for a manage-account url."""
249
250    weblocation: Annotated[WebLocation, IOAttrs('l')] = (
251        WebLocation.ACCOUNT_EDITOR
252    )
253
254    @override
255    @classmethod
256    def get_response_types(cls) -> list[type[Response] | None]:
257        return [ManageAccountResponse]
258
259
260@ioprepped
261@dataclass
262class ManageAccountResponse(Response):
263    """Here's that sign-in result you asked for, boss."""
264
265    url: Annotated[str | None, IOAttrs('u')]
266
267
268@ioprepped
269@dataclass
270class StoreQueryMessage(Message):
271    """Message asking about purchasable stuff and store related state."""
272
273    @override
274    @classmethod
275    def get_response_types(cls) -> list[type[Response] | None]:
276        return [StoreQueryResponse]
277
278
279@ioprepped
280@dataclass
281class StoreQueryResponse(Response):
282    """Here's that store info you asked for, boss."""
283
284    class Result(Enum):
285        """Our overall result."""
286
287        SUCCESS = 's'
288        ERROR = 'e'
289
290    @dataclass
291    class Purchase:
292        """Info about a purchasable thing."""
293
294        purchaseid: Annotated[str, IOAttrs('id')]
295
296    # Overall result; all data is undefined if not SUCCESS.
297    result: Annotated[Result, IOAttrs('r')]
298
299    tokens: Annotated[int, IOAttrs('t')]
300    gold_pass: Annotated[bool, IOAttrs('g')]
301
302    available_purchases: Annotated[list[Purchase], IOAttrs('p')]
303    token_info_url: Annotated[str, IOAttrs('tiu')]
304
305
306@ioprepped
307@dataclass
308class BSPrivatePartyMessage(Message):
309    """Message asking about info we need for private-party UI."""
310
311    need_datacode: Annotated[bool, IOAttrs('d')]
312
313    @override
314    @classmethod
315    def get_response_types(cls) -> list[type[Response] | None]:
316        return [BSPrivatePartyResponse]
317
318
319@ioprepped
320@dataclass
321class BSPrivatePartyResponse(Response):
322    """Here's that private party UI info you asked for, boss."""
323
324    success: Annotated[bool, IOAttrs('s')]
325    tokens: Annotated[int, IOAttrs('t')]
326    gold_pass: Annotated[bool, IOAttrs('g')]
327    datacode: Annotated[str | None, IOAttrs('d')]
328
329
330class BSClassicChestAppearance(Enum):
331    """Appearances bombsquad classic chests can have."""
332
333    UNKNOWN = 'u'
334    DEFAULT = 'd'
335
336
337@ioprepped
338@dataclass
339class BSClassicAccountLiveData:
340    """Account related data kept up to date live for classic app mode."""
341
342    @dataclass
343    class Chest:
344        """A lovely chest."""
345
346        appearance: Annotated[
347            BSClassicChestAppearance,
348            IOAttrs('a', enum_fallback=BSClassicChestAppearance.UNKNOWN),
349        ]
350        unlock_time: Annotated[datetime.datetime, IOAttrs('t')]
351        ad_allow_time: Annotated[datetime.datetime | None, IOAttrs('at')]
352
353    class LeagueType(Enum):
354        """Type of league we are in."""
355
356        BRONZE = 'b'
357        SILVER = 's'
358        GOLD = 'g'
359        DIAMOND = 'd'
360
361    tickets: Annotated[int, IOAttrs('ti')]
362
363    tokens: Annotated[int, IOAttrs('to')]
364    gold_pass: Annotated[bool, IOAttrs('g')]
365
366    achievements: Annotated[int, IOAttrs('a')]
367    achievements_total: Annotated[int, IOAttrs('at')]
368
369    league_type: Annotated[LeagueType | None, IOAttrs('lt')]
370    league_num: Annotated[int | None, IOAttrs('ln')]
371    league_rank: Annotated[int | None, IOAttrs('lr')]
372
373    level: Annotated[int, IOAttrs('lv')]
374    xp: Annotated[int, IOAttrs('xp')]
375    xpmax: Annotated[int, IOAttrs('xpm')]
376
377    inbox_count: Annotated[int, IOAttrs('ibc')]
378    inbox_count_is_max: Annotated[bool, IOAttrs('ibcm')]
379
380    chests: Annotated[dict[str, Chest], IOAttrs('c')]
381
382
383class BSInboxEntryType(Enum):
384    """Types of entries that can be in an inbox."""
385
386    UNKNOWN = 'u'  # Entry types we don't support will be this.
387    SIMPLE = 's'
388    CLAIM = 'c'
389    CLAIM_DISCARD = 'cd'
390
391
392@ioprepped
393@dataclass
394class BSInboxEntry:
395    """Single message in an inbox."""
396
397    type: Annotated[
398        BSInboxEntryType, IOAttrs('t', enum_fallback=BSInboxEntryType.UNKNOWN)
399    ]
400    id: Annotated[str, IOAttrs('i')]
401    createtime: Annotated[datetime.datetime, IOAttrs('c')]
402
403    # If clients don't support format_version of a message they will
404    # display 'app needs to be updated to show this'.
405    format_version: Annotated[int, IOAttrs('f', soft_default=1)]
406
407    # These have soft defaults so can be removed in the future if desired.
408    message: Annotated[str, IOAttrs('m', soft_default='(invalid message)')]
409    subs: Annotated[list[str], IOAttrs('s', soft_default_factory=list)]
410
411
412@ioprepped
413@dataclass
414class BSInboxRequestMessage(Message):
415    """Message requesting our inbox."""
416
417    @override
418    @classmethod
419    def get_response_types(cls) -> list[type[Response] | None]:
420        return [BSInboxRequestResponse]
421
422
423@ioprepped
424@dataclass
425class BSInboxRequestResponse(Response):
426    """Here's that inbox contents you asked for, boss."""
427
428    entries: Annotated[list[BSInboxEntry], IOAttrs('m')]
429
430    # Printable error if something goes wrong.
431    error: Annotated[str | None, IOAttrs('e')] = None
432
433
434@ioprepped
435@dataclass
436class BSChestInfoMessage(Message):
437    """Request info about a chest."""
438
439    chest_id: Annotated[str, IOAttrs('i')]
440
441    @override
442    @classmethod
443    def get_response_types(cls) -> list[type[Response] | None]:
444        return [BSChestInfoResponse]
445
446
447@ioprepped
448@dataclass
449class BSChestInfoResponse(Response):
450    """Here's that inbox contents you asked for, boss."""
451
452    @dataclass
453    class Chest:
454        """A lovely chest."""
455
456        appearance: Annotated[
457            BSClassicChestAppearance,
458            IOAttrs('a', enum_fallback=BSClassicChestAppearance.UNKNOWN),
459        ]
460
461        # How much to unlock *now*.
462        unlock_tokens: Annotated[int, IOAttrs('tk')]
463
464        # When unlocks on its own.
465        unlock_time: Annotated[datetime.datetime, IOAttrs('t')]
466
467        # Are ads allowed now?
468        ad_allow: Annotated[bool, IOAttrs('aa')]
469
470    chest: Annotated[Chest | None, IOAttrs('c')]
471
472
473@ioprepped
474@dataclass
475class BSChestActionMessage(Message):
476    """Request action about a chest."""
477
478    class Action(Enum):
479        """Types of actions we can request."""
480
481        # Unlocking (for free or with tokens).
482        UNLOCK = 'u'
483
484        # Watched an ad to reduce wait.
485        AD = 'ad'
486
487    action: Annotated[Action, IOAttrs('a')]
488
489    # Tokens we are paying (only applies to unlock).
490    token_payment: Annotated[int, IOAttrs('t')]
491
492    chest_id: Annotated[str, IOAttrs('i')]
493
494    @override
495    @classmethod
496    def get_response_types(cls) -> list[type[Response] | None]:
497        return [BSChestActionResponse]
498
499
500@ioprepped
501@dataclass
502class BSChestActionResponse(Response):
503    """Here's the results of that action you asked for, boss."""
504
505    # If present, signifies the chest has been opened and we should show
506    # the user this stuff that was in it.
507    contents: Annotated[list[str] | None, IOAttrs('c')] = None
508
509    # Printable error if something goes wrong.
510    error: Annotated[str | None, IOAttrs('e')] = None
511
512
513class BSInboxEntryProcessType(Enum):
514    """Types of processing we can ask for."""
515
516    POSITIVE = 'p'
517    NEGATIVE = 'n'
518
519
520@ioprepped
521@dataclass
522class BSInboxEntryProcessMessage(Message):
523    """Do something to an inbox entry."""
524
525    id: Annotated[str, IOAttrs('i')]
526    process_type: Annotated[BSInboxEntryProcessType, IOAttrs('t')]
527
528    @override
529    @classmethod
530    def get_response_types(cls) -> list[type[Response] | None]:
531        return [BSInboxEntryProcessResponse]
532
533
534@ioprepped
535@dataclass
536class BSInboxEntryProcessResponse(Response):
537    """Did something to that inbox entry, boss."""
538
539    # Printable error if something goes wrong.
540    error: Annotated[str | None, IOAttrs('e')] = None