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