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