matrix_sdk_base/store/send_queue.rs
1// Copyright 2024 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! All data types related to the send queue.
16
17use std::{collections::BTreeMap, fmt, ops::Deref};
18
19use as_variant::as_variant;
20use ruma::{
21 MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedEventId, OwnedTransactionId, OwnedUserId,
22 TransactionId, UInt,
23 events::{
24 AnyMessageLikeEventContent, MessageLikeEventContent as _, RawExt as _,
25 room::{MediaSource, message::RoomMessageEventContent},
26 },
27 serde::Raw,
28};
29use serde::{Deserialize, Serialize};
30
31use crate::media::MediaRequestParameters;
32
33/// A thin wrapper to serialize a `AnyMessageLikeEventContent`.
34#[derive(Clone, Serialize, Deserialize)]
35pub struct SerializableEventContent {
36 event: Raw<AnyMessageLikeEventContent>,
37 event_type: String,
38}
39
40#[cfg(not(tarpaulin_include))]
41impl fmt::Debug for SerializableEventContent {
42 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43 // Don't include the event in the debug display.
44 f.debug_struct("SerializedEventContent")
45 .field("event_type", &self.event_type)
46 .finish_non_exhaustive()
47 }
48}
49
50impl SerializableEventContent {
51 /// Create a [`SerializableEventContent`] from a raw
52 /// [`AnyMessageLikeEventContent`] along with its type.
53 pub fn from_raw(event: Raw<AnyMessageLikeEventContent>, event_type: String) -> Self {
54 Self { event_type, event }
55 }
56
57 /// Create a [`SerializableEventContent`] from an
58 /// [`AnyMessageLikeEventContent`].
59 pub fn new(event: &AnyMessageLikeEventContent) -> Result<Self, serde_json::Error> {
60 Ok(Self::from_raw(Raw::new(event)?, event.event_type().to_string()))
61 }
62
63 /// Convert a [`SerializableEventContent`] back into a
64 /// [`AnyMessageLikeEventContent`].
65 pub fn deserialize(&self) -> Result<AnyMessageLikeEventContent, serde_json::Error> {
66 self.event.deserialize_with_type(&self.event_type)
67 }
68
69 /// Returns the raw event content along with its type, borrowed variant.
70 ///
71 /// Useful for callers manipulating custom events.
72 pub fn raw(&self) -> (&Raw<AnyMessageLikeEventContent>, &str) {
73 (&self.event, &self.event_type)
74 }
75
76 /// Returns the raw event content along with its type, owned variant.
77 ///
78 /// Useful for callers manipulating custom events.
79 pub fn into_raw(self) -> (Raw<AnyMessageLikeEventContent>, String) {
80 (self.event, self.event_type)
81 }
82}
83
84/// The kind of a send queue request.
85#[derive(Clone, Debug, Serialize, Deserialize)]
86pub enum QueuedRequestKind {
87 /// An event to be sent via the send queue.
88 Event {
89 /// The content of the message-like event we'd like to send.
90 content: SerializableEventContent,
91 },
92
93 /// Content to upload on the media server.
94 ///
95 /// The bytes must be stored in the media cache, and are identified by the
96 /// cache key.
97 MediaUpload {
98 /// Content type of the media to be uploaded.
99 ///
100 /// Stored as a `String` because `Mime` which we'd really want to use
101 /// here, is not serializable. Oh well.
102 content_type: String,
103
104 /// The cache key used to retrieve the media's bytes in the event cache
105 /// store.
106 cache_key: MediaRequestParameters,
107
108 /// An optional media source for a thumbnail already uploaded.
109 thumbnail_source: Option<MediaSource>,
110
111 /// To which media event transaction does this upload relate?
112 related_to: OwnedTransactionId,
113
114 /// Accumulated list of infos for previously uploaded files and
115 /// thumbnails if used during a gallery transaction. Otherwise empty.
116 #[cfg(feature = "unstable-msc4274")]
117 #[serde(default)]
118 accumulated: Vec<AccumulatedSentMediaInfo>,
119 },
120}
121
122impl From<SerializableEventContent> for QueuedRequestKind {
123 fn from(content: SerializableEventContent) -> Self {
124 Self::Event { content }
125 }
126}
127
128/// A request to be sent with a send queue.
129#[derive(Clone)]
130pub struct QueuedRequest {
131 /// The kind of queued request we're going to send.
132 pub kind: QueuedRequestKind,
133
134 /// Unique transaction id for the queued request, acting as a key.
135 pub transaction_id: OwnedTransactionId,
136
137 /// Error returned when the request couldn't be sent and is stuck in the
138 /// unrecoverable state.
139 ///
140 /// `None` if the request is in the queue, waiting to be sent.
141 pub error: Option<QueueWedgeError>,
142
143 /// At which priority should this be handled?
144 ///
145 /// The bigger the value, the higher the priority at which this request
146 /// should be handled.
147 pub priority: usize,
148
149 /// The time that the request was originally attempted.
150 pub created_at: MilliSecondsSinceUnixEpoch,
151}
152
153impl QueuedRequest {
154 /// Returns `Some` if the queued request is about sending an event.
155 pub fn as_event(&self) -> Option<&SerializableEventContent> {
156 as_variant!(&self.kind, QueuedRequestKind::Event { content } => content)
157 }
158
159 /// True if the request couldn't be sent because of an unrecoverable API
160 /// error. See [`Self::error`] for more details on the reason.
161 pub fn is_wedged(&self) -> bool {
162 self.error.is_some()
163 }
164}
165
166/// Represents a failed to send unrecoverable error of an event sent via the
167/// send queue.
168///
169/// It is a serializable representation of a client error, see
170/// `From` implementation for more details. These errors can not be
171/// automatically retried, but yet some manual action can be taken before retry
172/// sending. If not the only solution is to delete the local event.
173#[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)]
174pub enum QueueWedgeError {
175 /// This error occurs when there are some insecure devices in the room, and
176 /// the current encryption setting prohibits sharing with them.
177 #[error("There are insecure devices in the room")]
178 InsecureDevices {
179 /// The insecure devices as a Map of userID to deviceID.
180 user_device_map: BTreeMap<OwnedUserId, Vec<OwnedDeviceId>>,
181 },
182
183 /// This error occurs when a previously verified user is not anymore, and
184 /// the current encryption setting prohibits sharing when it happens.
185 #[error("Some users that were previously verified are not anymore")]
186 IdentityViolations {
187 /// The users that are expected to be verified but are not.
188 users: Vec<OwnedUserId>,
189 },
190
191 /// It is required to set up cross-signing and properly verify the current
192 /// session before sending.
193 #[error("Own verification is required")]
194 CrossVerificationRequired,
195
196 /// Media content was cached in the media store, but has disappeared before
197 /// we could upload it.
198 #[error("Media content disappeared")]
199 MissingMediaContent,
200
201 /// We tried to upload some media content with an unknown mime type.
202 #[error("Invalid mime type '{mime_type}' for media")]
203 InvalidMimeType {
204 /// The observed mime type that's expected to be invalid.
205 mime_type: String,
206 },
207
208 /// Other errors.
209 #[error("Other unrecoverable error: {msg}")]
210 GenericApiError {
211 /// Description of the error.
212 msg: String,
213 },
214}
215
216/// The specific user intent that characterizes a [`DependentQueuedRequest`].
217#[derive(Clone, Debug, Serialize, Deserialize)]
218pub enum DependentQueuedRequestKind {
219 /// The event should be edited.
220 EditEvent {
221 /// The new event for the content.
222 new_content: SerializableEventContent,
223 },
224
225 /// The event should be redacted/aborted/removed.
226 RedactEvent,
227
228 /// The event should be reacted to, with the given key.
229 ReactEvent {
230 /// Key used for the reaction.
231 key: String,
232 },
233
234 /// Upload a file or thumbnail depending on another file or thumbnail
235 /// upload.
236 #[serde(alias = "UploadFileWithThumbnail")]
237 UploadFileOrThumbnail {
238 /// Content type for the file or thumbnail.
239 content_type: String,
240
241 /// Media request necessary to retrieve the file or thumbnail itself.
242 cache_key: MediaRequestParameters,
243
244 /// To which media transaction id does this upload relate to?
245 related_to: OwnedTransactionId,
246
247 /// Whether the depended upon request was a thumbnail or a file upload.
248 #[serde(default = "default_parent_is_thumbnail_upload")]
249 parent_is_thumbnail_upload: bool,
250 },
251
252 /// Finish an upload by updating references to the media cache and sending
253 /// the final media event with the remote MXC URIs.
254 FinishUpload {
255 /// Local echo for the event (containing the local MXC URIs).
256 ///
257 /// `Box` the local echo so that it reduces the size of the whole enum.
258 local_echo: Box<RoomMessageEventContent>,
259
260 /// Transaction id for the file upload.
261 file_upload: OwnedTransactionId,
262
263 /// Information about the thumbnail, if present.
264 thumbnail_info: Option<FinishUploadThumbnailInfo>,
265 },
266
267 /// Finish a gallery upload by updating references to the media cache and
268 /// sending the final gallery event with the remote MXC URIs.
269 #[cfg(feature = "unstable-msc4274")]
270 FinishGallery {
271 /// Local echo for the event (containing the local MXC URIs).
272 ///
273 /// `Box` the local echo so that it reduces the size of the whole enum.
274 local_echo: Box<RoomMessageEventContent>,
275
276 /// Metadata about the gallery items.
277 item_infos: Vec<FinishGalleryItemInfo>,
278 },
279}
280
281/// If parent_is_thumbnail_upload is missing, we assume the request is for a
282/// file upload following a thumbnail upload. This was the only possible case
283/// before parent_is_thumbnail_upload was introduced.
284fn default_parent_is_thumbnail_upload() -> bool {
285 true
286}
287
288/// Detailed record about a thumbnail used when finishing a media upload.
289#[derive(Clone, Debug, Serialize, Deserialize)]
290pub struct FinishUploadThumbnailInfo {
291 /// Transaction id for the thumbnail upload.
292 pub txn: OwnedTransactionId,
293 /// Thumbnail's width.
294 ///
295 /// Used previously, kept for backwards compatibility.
296 #[serde(default, skip_serializing_if = "Option::is_none")]
297 pub width: Option<UInt>,
298 /// Thumbnail's height.
299 ///
300 /// Used previously, kept for backwards compatibility.
301 #[serde(default, skip_serializing_if = "Option::is_none")]
302 pub height: Option<UInt>,
303}
304
305/// Detailed record about a file and thumbnail. When finishing a gallery
306/// upload, one [`FinishGalleryItemInfo`] will be used for each media in the
307/// gallery.
308#[cfg(feature = "unstable-msc4274")]
309#[derive(Clone, Debug, Serialize, Deserialize)]
310pub struct FinishGalleryItemInfo {
311 /// Transaction id for the file upload.
312 pub file_upload: OwnedTransactionId,
313 /// Information about the thumbnail, if present.
314 pub thumbnail_info: Option<FinishUploadThumbnailInfo>,
315}
316
317/// A transaction id identifying a [`DependentQueuedRequest`] rather than its
318/// parent [`QueuedRequest`].
319///
320/// This thin wrapper adds some safety to some APIs, making it possible to
321/// distinguish between the parent's `TransactionId` and the dependent event's
322/// own `TransactionId`.
323#[repr(transparent)]
324#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
325#[serde(transparent)]
326pub struct ChildTransactionId(OwnedTransactionId);
327
328impl ChildTransactionId {
329 /// Returns a new [`ChildTransactionId`].
330 #[allow(clippy::new_without_default)]
331 pub fn new() -> Self {
332 Self(TransactionId::new())
333 }
334}
335
336impl Deref for ChildTransactionId {
337 type Target = TransactionId;
338
339 fn deref(&self) -> &Self::Target {
340 &self.0
341 }
342}
343
344impl From<String> for ChildTransactionId {
345 fn from(val: String) -> Self {
346 Self(val.into())
347 }
348}
349
350impl From<ChildTransactionId> for OwnedTransactionId {
351 fn from(val: ChildTransactionId) -> Self {
352 val.0
353 }
354}
355
356impl From<OwnedTransactionId> for ChildTransactionId {
357 fn from(val: OwnedTransactionId) -> Self {
358 Self(val)
359 }
360}
361
362/// Information about a media (and its thumbnail) that have been sent to a
363/// homeserver.
364#[derive(Clone, Debug, Serialize, Deserialize)]
365pub struct SentMediaInfo {
366 /// File that was uploaded by this request.
367 ///
368 /// If the request related to a thumbnail upload, this contains the
369 /// thumbnail media source.
370 pub file: MediaSource,
371
372 /// Optional thumbnail previously uploaded, when uploading a file.
373 ///
374 /// When uploading a thumbnail, this is set to `None`.
375 pub thumbnail: Option<MediaSource>,
376
377 /// Accumulated list of infos for previously uploaded files and thumbnails
378 /// if used during a gallery transaction. Otherwise empty.
379 #[cfg(feature = "unstable-msc4274")]
380 #[serde(default)]
381 pub accumulated: Vec<AccumulatedSentMediaInfo>,
382}
383
384/// Accumulated information about a media (and its thumbnail) that have been
385/// sent to a homeserver.
386#[cfg(feature = "unstable-msc4274")]
387#[derive(Clone, Debug, Serialize, Deserialize)]
388pub struct AccumulatedSentMediaInfo {
389 /// File that was uploaded by this request.
390 ///
391 /// If the request related to a thumbnail upload, this contains the
392 /// thumbnail media source.
393 pub file: MediaSource,
394
395 /// Optional thumbnail previously uploaded, when uploading a file.
396 ///
397 /// When uploading a thumbnail, this is set to `None`.
398 pub thumbnail: Option<MediaSource>,
399}
400
401#[cfg(feature = "unstable-msc4274")]
402impl From<AccumulatedSentMediaInfo> for SentMediaInfo {
403 fn from(value: AccumulatedSentMediaInfo) -> Self {
404 Self { file: value.file, thumbnail: value.thumbnail, accumulated: vec![] }
405 }
406}
407
408/// A unique key (identifier) indicating that a transaction has been
409/// successfully sent to the server.
410///
411/// The owning child transactions can now be resolved.
412#[derive(Clone, Debug, Serialize, Deserialize)]
413pub enum SentRequestKey {
414 /// The parent transaction returned an event when it succeeded.
415 Event(OwnedEventId),
416
417 /// The parent transaction returned an uploaded resource URL.
418 Media(SentMediaInfo),
419}
420
421impl SentRequestKey {
422 /// Converts the current parent key into an event id, if possible.
423 pub fn into_event_id(self) -> Option<OwnedEventId> {
424 as_variant!(self, Self::Event)
425 }
426
427 /// Converts the current parent key into information about a sent media, if
428 /// possible.
429 pub fn into_media(self) -> Option<SentMediaInfo> {
430 as_variant!(self, Self::Media)
431 }
432}
433
434/// A request to be sent, depending on a [`QueuedRequest`] to be sent first.
435///
436/// Depending on whether the parent request has been sent or not, this will
437/// either update the local echo in the storage, or materialize an equivalent
438/// request implementing the user intent to the homeserver.
439#[derive(Clone, Debug, Serialize, Deserialize)]
440pub struct DependentQueuedRequest {
441 /// Unique identifier for this dependent queued request.
442 ///
443 /// Useful for deletion.
444 pub own_transaction_id: ChildTransactionId,
445
446 /// The kind of user intent.
447 pub kind: DependentQueuedRequestKind,
448
449 /// Transaction id for the parent's local echo / used in the server request.
450 ///
451 /// Note: this is the transaction id used for the depended-on request, i.e.
452 /// the one that was originally sent and that's being modified with this
453 /// dependent request.
454 pub parent_transaction_id: OwnedTransactionId,
455
456 /// If the parent request has been sent, the parent's request identifier
457 /// returned by the server once the local echo has been sent out.
458 pub parent_key: Option<SentRequestKey>,
459
460 /// The time that the request was originally attempted.
461 pub created_at: MilliSecondsSinceUnixEpoch,
462}
463
464impl DependentQueuedRequest {
465 /// Does the dependent request represent a new event that is *not*
466 /// aggregated, aka it is going to be its own item in a timeline?
467 pub fn is_own_event(&self) -> bool {
468 match self.kind {
469 DependentQueuedRequestKind::EditEvent { .. }
470 | DependentQueuedRequestKind::RedactEvent
471 | DependentQueuedRequestKind::ReactEvent { .. }
472 | DependentQueuedRequestKind::UploadFileOrThumbnail { .. } => {
473 // These are all aggregated events, or non-visible items (file upload producing
474 // a new MXC ID).
475 false
476 }
477 DependentQueuedRequestKind::FinishUpload { .. } => {
478 // This one graduates into a new media event.
479 true
480 }
481 #[cfg(feature = "unstable-msc4274")]
482 DependentQueuedRequestKind::FinishGallery { .. } => {
483 // This one graduates into a new gallery event.
484 true
485 }
486 }
487 }
488}
489
490#[cfg(not(tarpaulin_include))]
491impl fmt::Debug for QueuedRequest {
492 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
493 // Hide the content from the debug log.
494 f.debug_struct("QueuedRequest")
495 .field("transaction_id", &self.transaction_id)
496 .field("is_wedged", &self.is_wedged())
497 .finish_non_exhaustive()
498 }
499}