matrix_sdk/room/
futures.rs

1// Copyright 2023 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//! Named futures returned from methods on types in [the `room` module][super].
16
17#![deny(unreachable_pub)]
18
19#[cfg(feature = "experimental-encrypted-state-events")]
20use std::borrow::Borrow;
21use std::future::IntoFuture;
22#[cfg(feature = "unstable-msc4354")]
23use std::time::Duration;
24
25use eyeball::SharedObservable;
26use matrix_sdk_base::deserialized_responses::EncryptionInfo;
27use matrix_sdk_common::boxed_into_future;
28use mime::Mime;
29#[cfg(feature = "e2e-encryption")]
30use ruma::events::MessageLikeEventType;
31#[cfg(feature = "unstable-msc4354")]
32use ruma::events::sticky::StickyDurationMs;
33#[cfg(doc)]
34use ruma::events::{MessageLikeUnsigned, SyncMessageLikeEvent};
35#[cfg(feature = "experimental-encrypted-state-events")]
36use ruma::events::{StateEventContent, StateEventType};
37use ruma::{
38    OwnedTransactionId, TransactionId,
39    api::client::{message::send_message_event, state::send_state_event},
40    assign,
41    events::{AnyMessageLikeEventContent, AnyStateEventContent, MessageLikeEventContent},
42    serde::Raw,
43};
44use tracing::{Instrument, Span, info, trace};
45
46use super::Room;
47use crate::{
48    Result, TransmissionProgress,
49    attachment::AttachmentConfig,
50    config::RequestConfig,
51    utils::{IntoRawMessageLikeEventContent, IntoRawStateEventContent},
52};
53
54/// The result of the [`Room::send`] future
55#[derive(Debug)]
56pub struct SendMessageLikeEventResult {
57    /// The response
58    pub response: send_message_event::v3::Response,
59    /// The encryption info, if the event was encrypted
60    pub encryption_info: Option<EncryptionInfo>,
61}
62
63/// Future returned by [`Room::send`].
64#[allow(missing_debug_implementations)]
65pub struct SendMessageLikeEvent<'a> {
66    room: &'a Room,
67    event_type: String,
68    content: serde_json::Result<serde_json::Value>,
69    transaction_id: Option<OwnedTransactionId>,
70    request_config: Option<RequestConfig>,
71    #[cfg(feature = "unstable-msc4354")]
72    sticky_duration: Option<StickyDurationMs>,
73}
74
75impl<'a> SendMessageLikeEvent<'a> {
76    pub(crate) fn new(room: &'a Room, content: impl MessageLikeEventContent) -> Self {
77        let event_type = content.event_type().to_string();
78        let content = serde_json::to_value(&content);
79        Self {
80            room,
81            event_type,
82            content,
83            transaction_id: None,
84            request_config: None,
85            #[cfg(feature = "unstable-msc4354")]
86            sticky_duration: None,
87        }
88    }
89
90    /// Make this event sticky for `duration`, clamped to one hour.
91    ///
92    /// Note that if the homeserver doesn't support sticky events, it will
93    /// ignore the duration and send the event unsticky. Server support can
94    /// be checked with [`Client::supports_sticky_events`].
95    ///
96    /// [`Client::supports_sticky_events`]: crate::Client::supports_sticky_events
97    #[cfg(feature = "unstable-msc4354")]
98    pub fn with_sticky_duration(mut self, duration: Duration) -> Self {
99        self.sticky_duration = Some(crate::utils::sticky_duration_ms(duration));
100        self
101    }
102
103    /// Set a transaction ID for this event.
104    ///
105    /// Since sending message-like events always requires a transaction ID, one
106    /// is generated if this method is not called.
107    ///
108    /// The transaction ID is a locally-unique ID describing a message
109    /// transaction with the homeserver.
110    ///
111    /// - On the sending side, this field is used for re-trying earlier failed
112    ///   transactions. Subsequent messages _must never_ re-use an earlier
113    ///   transaction ID.
114    /// - On the receiving side, the field is used for recognizing our own
115    ///   messages when they arrive down the sync: the server includes the ID in
116    ///   the [`MessageLikeUnsigned`] field `transaction_id` of the
117    ///   corresponding [`SyncMessageLikeEvent`], but only for the _sending_
118    ///   device. Other devices will not see it. This is then used to ignore
119    ///   events sent by our own device and/or to implement local echo.
120    pub fn with_transaction_id(mut self, txn_id: OwnedTransactionId) -> Self {
121        self.transaction_id = Some(txn_id);
122        self
123    }
124
125    /// Assign a given [`RequestConfig`] to configure how this request should
126    /// behave with respect to the network.
127    pub fn with_request_config(mut self, request_config: RequestConfig) -> Self {
128        self.request_config = Some(request_config);
129        self
130    }
131}
132
133impl<'a> IntoFuture for SendMessageLikeEvent<'a> {
134    type Output = Result<SendMessageLikeEventResult>;
135    boxed_into_future!(extra_bounds: 'a);
136
137    fn into_future(self) -> Self::IntoFuture {
138        let Self {
139            room,
140            event_type,
141            content,
142            transaction_id,
143            request_config,
144            #[cfg(feature = "unstable-msc4354")]
145            sticky_duration,
146        } = self;
147        Box::pin(async move {
148            let content = content?;
149            let future =
150                assign!(room.send_raw(&event_type, content), { transaction_id, request_config });
151            #[cfg(feature = "unstable-msc4354")]
152            let future = assign!(future, { sticky_duration });
153            future.await
154        })
155    }
156}
157
158/// Future returned by [`Room::send_raw`].
159#[allow(missing_debug_implementations)]
160pub struct SendRawMessageLikeEvent<'a> {
161    room: &'a Room,
162    event_type: &'a str,
163    content: Raw<AnyMessageLikeEventContent>,
164    tracing_span: Span,
165    transaction_id: Option<OwnedTransactionId>,
166    request_config: Option<RequestConfig>,
167    #[cfg(feature = "unstable-msc4354")]
168    sticky_duration: Option<StickyDurationMs>,
169}
170
171impl<'a> SendRawMessageLikeEvent<'a> {
172    pub(crate) fn new(
173        room: &'a Room,
174        event_type: &'a str,
175        content: impl IntoRawMessageLikeEventContent,
176    ) -> Self {
177        let content = content.into_raw_message_like_event_content();
178        Self {
179            room,
180            event_type,
181            content,
182            tracing_span: Span::current(),
183            transaction_id: None,
184            request_config: None,
185            #[cfg(feature = "unstable-msc4354")]
186            sticky_duration: None,
187        }
188    }
189
190    /// Make this event sticky for `duration`, clamped to one hour.
191    ///
192    /// Note that if the homeserver doesn't support sticky events, it will
193    /// ignore the duration and send the event unsticky. Server support can
194    /// be checked with [`Client::supports_sticky_events`].
195    ///
196    /// [`Client::supports_sticky_events`]: crate::Client::supports_sticky_events
197    #[cfg(feature = "unstable-msc4354")]
198    pub fn with_sticky_duration(mut self, duration: Duration) -> Self {
199        self.sticky_duration = Some(crate::utils::sticky_duration_ms(duration));
200        self
201    }
202
203    /// Set a transaction ID for this event.
204    ///
205    /// Since sending message-like events always requires a transaction ID, one
206    /// is generated if this method is not called.
207    ///
208    /// - On the sending side, this field is used for re-trying earlier failed
209    ///   transactions. Subsequent messages _must never_ re-use an earlier
210    ///   transaction ID.
211    /// - On the receiving side, the field is used for recognizing our own
212    ///   messages when they arrive down the sync: the server includes the ID in
213    ///   the [`MessageLikeUnsigned`] field `transaction_id` of the
214    ///   corresponding [`SyncMessageLikeEvent`], but only for the _sending_
215    ///   device. Other devices will not see it. This is then used to ignore
216    ///   events sent by our own device and/or to implement local echo.
217    pub fn with_transaction_id(mut self, txn_id: &TransactionId) -> Self {
218        self.transaction_id = Some(txn_id.to_owned());
219        self
220    }
221
222    /// Assign a given [`RequestConfig`] to configure how this request should
223    /// behave with respect to the network.
224    pub fn with_request_config(mut self, request_config: RequestConfig) -> Self {
225        self.request_config = Some(request_config);
226        self
227    }
228}
229
230impl<'a> IntoFuture for SendRawMessageLikeEvent<'a> {
231    type Output = Result<SendMessageLikeEventResult>;
232    boxed_into_future!(extra_bounds: 'a);
233
234    fn into_future(self) -> Self::IntoFuture {
235        #[cfg_attr(not(feature = "e2e-encryption"), allow(unused_mut))]
236        let Self {
237            room,
238            mut event_type,
239            mut content,
240            tracing_span,
241            transaction_id,
242            request_config,
243            #[cfg(feature = "unstable-msc4354")]
244            sticky_duration,
245        } = self;
246
247        let fut = async move {
248            room.ensure_room_joined()?;
249
250            let txn_id = transaction_id.unwrap_or_else(TransactionId::new);
251            Span::current().record("transaction_id", tracing::field::debug(&txn_id));
252
253            #[cfg(not(feature = "e2e-encryption"))]
254            trace!("Sending plaintext event to room because we don't have encryption support.");
255
256            #[cfg(feature = "e2e-encryption")]
257            let mut encryption_info: Option<EncryptionInfo> = None;
258            #[cfg(not(feature = "e2e-encryption"))]
259            let encryption_info: Option<EncryptionInfo> = None;
260
261            #[cfg(feature = "e2e-encryption")]
262            if room.latest_encryption_state().await?.is_encrypted() {
263                Span::current().record("is_room_encrypted", true);
264                // Reactions are currently famously not encrypted, skip
265                // encrypting them until they are.
266                if MessageLikeEventType::from(event_type) == MessageLikeEventType::Reaction {
267                    trace!("Sending plaintext event because of the event type.");
268                } else {
269                    trace!(
270                        room_id = ?room.room_id(),
271                        "Sending encrypted event because the room is encrypted.",
272                    );
273
274                    ensure_room_encryption_ready(room).await?;
275
276                    let olm = room.client.olm_machine().await;
277                    let olm = olm.as_ref().expect("Olm machine wasn't started");
278
279                    let result =
280                        olm.encrypt_room_event_raw(room.room_id(), event_type, &content).await?;
281                    content = result.content.cast();
282                    encryption_info = Some(result.encryption_info);
283                    event_type = "m.room.encrypted";
284                }
285            } else {
286                Span::current().record("is_room_encrypted", false);
287                trace!("Sending plaintext event because the room is NOT encrypted.");
288            }
289
290            let request = send_message_event::v3::Request::new_raw(
291                room.room_id().to_owned(),
292                txn_id,
293                event_type.into(),
294                content,
295            );
296            #[cfg(feature = "unstable-msc4354")]
297            let request = assign!(request, { sticky_duration_ms: sticky_duration });
298
299            let response = room.client.send(request).with_request_config(request_config).await?;
300
301            Span::current().record("event_id", tracing::field::debug(&response.event_id));
302            info!("Sent event in room");
303
304            Ok(SendMessageLikeEventResult { response, encryption_info })
305        };
306
307        Box::pin(fut.instrument(tracing_span))
308    }
309}
310
311/// Future returned by [`Room::send_attachment`].
312#[allow(missing_debug_implementations)]
313pub struct SendAttachment<'a> {
314    room: &'a Room,
315    filename: String,
316    content_type: &'a Mime,
317    data: Vec<u8>,
318    config: AttachmentConfig,
319    tracing_span: Span,
320    send_progress: SharedObservable<TransmissionProgress>,
321    store_in_cache: bool,
322}
323
324impl<'a> SendAttachment<'a> {
325    pub(crate) fn new(
326        room: &'a Room,
327        filename: String,
328        content_type: &'a Mime,
329        data: Vec<u8>,
330        config: AttachmentConfig,
331    ) -> Self {
332        Self {
333            room,
334            filename,
335            content_type,
336            data,
337            config,
338            tracing_span: Span::current(),
339            send_progress: Default::default(),
340            store_in_cache: false,
341        }
342    }
343
344    /// Replace the default `SharedObservable` used for tracking upload
345    /// progress.
346    pub fn with_send_progress_observable(
347        mut self,
348        send_progress: SharedObservable<TransmissionProgress>,
349    ) -> Self {
350        self.send_progress = send_progress;
351        self
352    }
353
354    /// Whether the sent attachment should be stored in the cache or not.
355    ///
356    /// If set to true, then retrieving the data for the attachment will result
357    /// in a cache hit immediately after upload.
358    pub fn store_in_cache(mut self) -> Self {
359        self.store_in_cache = true;
360        self
361    }
362}
363
364impl<'a> IntoFuture for SendAttachment<'a> {
365    type Output = Result<send_message_event::v3::Response>;
366    boxed_into_future!(extra_bounds: 'a);
367
368    fn into_future(self) -> Self::IntoFuture {
369        let Self {
370            room,
371            filename,
372            content_type,
373            data,
374            config,
375            tracing_span,
376            send_progress,
377            store_in_cache,
378        } = self;
379        let fut = async move {
380            room.prepare_and_send_attachment(
381                filename,
382                content_type,
383                data,
384                config,
385                send_progress,
386                store_in_cache,
387            )
388            .await
389        };
390
391        Box::pin(fut.instrument(tracing_span))
392    }
393}
394
395/// Future returned by [`Room::send_state_event_raw`].
396#[allow(missing_debug_implementations)]
397pub struct SendRawStateEvent<'a> {
398    room: &'a Room,
399    event_type: &'a str,
400    state_key: &'a str,
401    content: Raw<AnyStateEventContent>,
402    tracing_span: Span,
403    request_config: Option<RequestConfig>,
404}
405
406impl<'a> SendRawStateEvent<'a> {
407    pub(crate) fn new(
408        room: &'a Room,
409        event_type: &'a str,
410        state_key: &'a str,
411        content: impl IntoRawStateEventContent,
412    ) -> Self {
413        let content = content.into_raw_state_event_content();
414        Self {
415            room,
416            event_type,
417            state_key,
418            content,
419            tracing_span: Span::current(),
420            request_config: None,
421        }
422    }
423
424    /// Assign a given [`RequestConfig`] to configure how this request should
425    /// behave with respect to the network.
426    pub fn with_request_config(mut self, request_config: RequestConfig) -> Self {
427        self.request_config = Some(request_config);
428        self
429    }
430
431    /// Determines whether the inner state event should be encrypted before
432    /// sending.
433    ///
434    /// This method checks two conditions:
435    ///
436    /// 1. Whether the room supports encrypted state events, by inspecting the
437    ///    room's encryption state.
438    /// 2. Whether the event type is considered "critical" or excluded from
439    ///    encryption under MSC4362.
440    ///
441    /// # Returns
442    ///
443    /// Returns `true` if the event should be encrypted, otherwise returns
444    /// `false`.
445    #[cfg(feature = "experimental-encrypted-state-events")]
446    fn should_encrypt(room: &Room, event_type: &str) -> bool {
447        if !room.encryption_state().is_state_encrypted() {
448            trace!("Sending plaintext event as the room does NOT support encrypted state events.");
449            return false;
450        }
451
452        // Check the event is not critical.
453        if matches!(
454            StateEventType::from(event_type),
455            StateEventType::RoomCreate
456                | StateEventType::RoomMember
457                | StateEventType::RoomJoinRules
458                | StateEventType::RoomPowerLevels
459                | StateEventType::RoomThirdPartyInvite
460                | StateEventType::RoomHistoryVisibility
461                | StateEventType::RoomGuestAccess
462                | StateEventType::RoomEncryption
463                | StateEventType::SpaceChild
464                | StateEventType::SpaceParent
465        ) {
466            trace!("Sending plaintext event as its type is excluded from encryption.");
467            return false;
468        }
469
470        true
471    }
472}
473
474impl<'a> IntoFuture for SendRawStateEvent<'a> {
475    type Output = Result<send_state_event::v3::Response>;
476    boxed_into_future!(extra_bounds: 'a);
477
478    fn into_future(self) -> Self::IntoFuture {
479        #[cfg_attr(not(feature = "experimental-encrypted-state-events"), allow(unused_mut))]
480        let Self { room, mut event_type, state_key, mut content, tracing_span, request_config } =
481            self;
482
483        let fut = async move {
484            room.ensure_room_joined()?;
485
486            #[cfg_attr(not(feature = "experimental-encrypted-state-events"), allow(unused_mut))]
487            let mut state_key = state_key.to_owned();
488
489            #[cfg(feature = "experimental-encrypted-state-events")]
490            if Self::should_encrypt(room, event_type) {
491                use tracing::debug;
492
493                Span::current().record("should_encrypt", true);
494                debug!(
495                    room_id = ?room.room_id(),
496                    "Sending encrypted event because the room is encrypted.",
497                );
498
499                ensure_room_encryption_ready(room).await?;
500
501                let olm = room.client.olm_machine().await;
502                let olm = olm.as_ref().expect("Olm machine wasn't started");
503
504                content = olm
505                    .encrypt_state_event_raw(room.room_id(), event_type, &state_key, &content)
506                    .await?
507                    .cast_unchecked();
508
509                state_key = format!("{event_type}:{state_key}");
510                event_type = "m.room.encrypted";
511            } else {
512                Span::current().record("should_encrypt", false);
513            }
514
515            let request = send_state_event::v3::Request::new_raw(
516                room.room_id().to_owned(),
517                event_type.into(),
518                state_key,
519                content,
520            );
521
522            let response = room.client.send(request).with_request_config(request_config).await?;
523
524            Span::current().record("event_id", tracing::field::debug(&response.event_id));
525            info!("Sent event in room");
526
527            Ok(response)
528        };
529
530        Box::pin(fut.instrument(tracing_span))
531    }
532}
533
534/// Future returned by `Room::send_state_event`.
535#[allow(missing_debug_implementations)]
536#[cfg(feature = "experimental-encrypted-state-events")]
537pub struct SendStateEvent<'a> {
538    room: &'a Room,
539    event_type: String,
540    state_key: String,
541    content: serde_json::Result<serde_json::Value>,
542    request_config: Option<RequestConfig>,
543}
544
545#[cfg(feature = "experimental-encrypted-state-events")]
546impl<'a> SendStateEvent<'a> {
547    pub(crate) fn new<C, K>(room: &'a Room, state_key: &K, content: C) -> Self
548    where
549        C: StateEventContent,
550        C::StateKey: Borrow<K>,
551        K: AsRef<str> + ?Sized,
552    {
553        let event_type = content.event_type().to_string();
554        let state_key = state_key.as_ref().to_owned();
555        let content = serde_json::to_value(&content);
556        Self { room, event_type, state_key, content, request_config: None }
557    }
558
559    /// Assign a given [`RequestConfig`] to configure how this request should
560    /// behave with respect to the network.
561    pub fn with_request_config(mut self, request_config: RequestConfig) -> Self {
562        self.request_config = Some(request_config);
563        self
564    }
565}
566
567#[cfg(feature = "experimental-encrypted-state-events")]
568impl<'a> IntoFuture for SendStateEvent<'a> {
569    type Output = Result<send_state_event::v3::Response>;
570    boxed_into_future!(extra_bounds: 'a);
571
572    fn into_future(self) -> Self::IntoFuture {
573        let Self { room, state_key, event_type, content, request_config } = self;
574        Box::pin(async move {
575            let content = content?;
576            assign!(room.send_state_event_raw(&event_type, &state_key, content), { request_config })
577                .await
578        })
579    }
580}
581
582/// Ensures the room is ready for encrypted events to be sent.
583#[cfg(feature = "e2e-encryption")]
584async fn ensure_room_encryption_ready(room: &Room) -> Result<()> {
585    if !room.are_members_synced() {
586        room.sync_members().await?;
587    }
588
589    // Query keys in case we don't have them for newly synced members.
590    //
591    // Note we do it all the time, because we might have sync'd members before
592    // sending a message (so didn't enter the above branch), but could have not
593    // query their keys ever.
594    room.query_keys_for_untracked_or_dirty_users().await?;
595
596    room.preshare_room_key().await?;
597
598    Ok(())
599}
matrix_sdk/room/futures.rs

matrix_sdk/room/
futures.rs
