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

matrix_sdk/room/
futures.rs
