matrix_sdk/

media.rs

// Copyright 2021 Kévin Commaille
// Copyright 2022 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! High-level media API.

#[cfg(feature = "e2e-encryption")]
use std::io::Read;
use std::{fmt, time::Duration};
#[cfg(not(target_family = "wasm"))]
use std::{fs::File, path::Path};

use eyeball::SharedObservable;
use futures_util::future::try_join;
use matrix_sdk_base::media::store::IgnoreMediaRetentionPolicy;
pub use matrix_sdk_base::media::{store::MediaRetentionPolicy, *};
use matrix_sdk_common::{BoxFuture, SendOutsideWasm, SyncOutsideWasm};
use mime::Mime;
use ruma::{
    MilliSecondsSinceUnixEpoch, MxcUri, OwnedMxcUri, TransactionId, UInt,
    api::{
        Metadata,
        client::{authenticated_media, media},
        error::ErrorKind,
    },
    assign,
    events::room::{MediaSource, ThumbnailInfo},
};
use serde_json::value::RawValue as RawJsonValue;
#[cfg(not(target_family = "wasm"))]
use tempfile::{Builder as TempFileBuilder, NamedTempFile, TempDir};
#[cfg(not(target_family = "wasm"))]
use tokio::{fs::File as TokioFile, io::AsyncWriteExt};

use crate::{
    Client, Error, Result, TransmissionProgress, attachment::Thumbnail,
    client::futures::SendMediaUploadRequest, config::RequestConfig,
};

/// A conservative upload speed of 1Mbps
const DEFAULT_UPLOAD_SPEED: u64 = 125_000;
/// 5 min minimal upload request timeout, used to clamp the request timeout.
const MIN_UPLOAD_REQUEST_TIMEOUT: Duration = Duration::from_secs(60 * 5);
/// The server name used to generate local MXC URIs.
// This mustn't represent a potentially valid media server, otherwise it'd be
// possible for an attacker to return malicious content under some
// preconditions (e.g. the cache store has been cleared before the upload
// took place). To mitigate against this, we use the .localhost TLD,
// which is guaranteed to be on the local machine. As a result, the only attack
// possible would be coming from the user themselves, which we consider a
// non-threat.
const LOCAL_MXC_SERVER_NAME: &str = "send-queue.localhost";

/// A high-level API to interact with the media API.
#[derive(Debug, Clone)]
pub struct Media {
    /// The underlying HTTP client.
    client: Client,
}

/// A file handle that takes ownership of a media file on disk. When the handle
/// is dropped, the file will be removed from the disk.
#[derive(Debug)]
#[cfg(not(target_family = "wasm"))]
pub struct MediaFileHandle {
    /// The temporary file that contains the media.
    file: NamedTempFile,
    /// An intermediary temporary directory used in certain cases.
    ///
    /// Only stored for its `Drop` semantics.
    _directory: Option<TempDir>,
}

#[cfg(not(target_family = "wasm"))]
impl MediaFileHandle {
    /// Get the media file's path.
    pub fn path(&self) -> &Path {
        self.file.path()
    }

    /// Persist the media file to the given path.
    pub fn persist(self, path: &Path) -> Result<File, PersistError> {
        self.file.persist(path).map_err(|e| PersistError {
            error: e.error,
            file: Self { file: e.file, _directory: self._directory },
        })
    }
}

/// Error returned when [`MediaFileHandle::persist`] fails.
#[cfg(not(target_family = "wasm"))]
pub struct PersistError {
    /// The underlying IO error.
    pub error: std::io::Error,
    /// The temporary file that couldn't be persisted.
    pub file: MediaFileHandle,
}

#[cfg(not(any(target_family = "wasm", tarpaulin_include)))]
impl fmt::Debug for PersistError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "PersistError({:?})", self.error)
    }
}

#[cfg(not(any(target_family = "wasm", tarpaulin_include)))]
impl fmt::Display for PersistError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "failed to persist temporary file: {}", self.error)
    }
}

/// A preallocated MXC URI created by [`Media::create_content_uri()`], and
/// to be used with [`Media::upload_preallocated()`].
#[derive(Debug)]
pub struct PreallocatedMxcUri {
    /// The URI for the media URI.
    pub uri: OwnedMxcUri,
    /// The expiration date for the media URI.
    expire_date: Option<MilliSecondsSinceUnixEpoch>,
}

/// An error that happened in the realm of media.
#[derive(Debug, thiserror::Error)]
pub enum MediaError {
    /// A preallocated MXC URI has expired.
    #[error("a preallocated MXC URI has expired")]
    ExpiredPreallocatedMxcUri,

    /// Preallocated media already had content, cannot overwrite.
    #[error("preallocated media already had content, cannot overwrite")]
    CannotOverwriteMedia,

    /// Local-only media content was not found.
    #[error("local-only media content was not found")]
    LocalMediaNotFound,

    /// The provided media is too large to upload.
    #[error(
        "The provided media is too large to upload. \
         Maximum upload length is {max} bytes, tried to upload {current} bytes"
    )]
    MediaTooLargeToUpload {
        /// The `max_upload_size` value for this homeserver.
        max: UInt,
        /// The size of the current media to upload.
        current: UInt,
    },

    /// Fetching the `max_upload_size` value from the homeserver failed.
    #[error("Fetching the `max_upload_size` value from the homeserver failed: {0}")]
    FetchMaxUploadSizeFailed(String),
}

/// A generic trait for fetching media content.
pub trait MediaFetcher: SendOutsideWasm + SyncOutsideWasm + fmt::Debug {
    /// Fetches the media content for the given [`MediaRequestParameters`].
    /// Returns either a byte array or an [`crate::Error`].
    fn fetch_media_content<'a>(
        &'a self,
        client: &'a Client,
        request: &'a MediaRequestParameters,
    ) -> BoxFuture<'a, Result<Vec<u8>, Error>>;
}

impl Media {
    pub(crate) fn new(client: Client) -> Self {
        Self { client }
    }

    /// Upload some media to the server.
    ///
    /// # Arguments
    ///
    /// * `content_type` - The type of the media, this will be used as the
    ///   content-type header.
    ///
    /// * `data` - Vector of bytes to be uploaded to the server.
    ///
    /// * `request_config` - Optional request configuration for the HTTP client,
    ///   overriding the default. If not provided, a reasonable timeout value is
    ///   inferred.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use std::fs;
    /// # use matrix_sdk::{Client, ruma::room_id};
    /// # use url::Url;
    /// # use mime;
    /// # async {
    /// # let homeserver = Url::parse("http://localhost:8080")?;
    /// # let mut client = Client::new(homeserver).await?;
    /// let image = fs::read("/home/example/my-cat.jpg")?;
    ///
    /// let response =
    ///     client.media().upload(&mime::IMAGE_JPEG, image, None).await?;
    ///
    /// println!("Cat URI: {}", response.content_uri);
    /// # anyhow::Ok(()) };
    /// ```
    pub fn upload(
        &self,
        content_type: &Mime,
        data: Vec<u8>,
        request_config: Option<RequestConfig>,
    ) -> SendMediaUploadRequest {
        let request_config = request_config.unwrap_or_else(|| {
            self.client.request_config().timeout(Self::reasonable_upload_timeout(&data))
        });

        let request = assign!(media::create_content::v3::Request::new(data), {
            content_type: Some(content_type.essence_str().to_owned()),
        });

        let request = self.client.send(request).with_request_config(request_config);
        SendMediaUploadRequest::new(request)
    }

    /// Returns a reasonable upload timeout for an upload, based on the size of
    /// the data to be uploaded.
    pub(crate) fn reasonable_upload_timeout(data: &[u8]) -> Duration {
        std::cmp::max(
            Duration::from_secs(data.len() as u64 / DEFAULT_UPLOAD_SPEED),
            MIN_UPLOAD_REQUEST_TIMEOUT,
        )
    }

    /// Preallocates an MXC URI for a media that will be uploaded soon.
    ///
    /// This preallocates an URI *before* any content is uploaded to the server.
    /// The resulting preallocated MXC URI can then be consumed with
    /// [`Media::upload_preallocated`].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use std::fs;
    /// # use matrix_sdk::{Client, ruma::room_id};
    /// # use url::Url;
    /// # use mime;
    /// # async {
    /// # let homeserver = Url::parse("http://localhost:8080")?;
    /// # let mut client = Client::new(homeserver).await?;
    ///
    /// let preallocated = client.media().create_content_uri().await?;
    /// println!("Cat URI: {}", preallocated.uri);
    ///
    /// let image = fs::read("/home/example/my-cat.jpg")?;
    /// client
    ///     .media()
    ///     .upload_preallocated(preallocated, &mime::IMAGE_JPEG, image)
    ///     .await?;
    ///
    /// # anyhow::Ok(()) };
    /// ```
    pub async fn create_content_uri(&self) -> Result<PreallocatedMxcUri> {
        // Note: this request doesn't have any parameters.
        let request = media::create_mxc_uri::v1::Request::default();

        let response = self.client.send(request).await?;

        Ok(PreallocatedMxcUri {
            uri: response.content_uri,
            expire_date: response.unused_expires_at,
        })
    }

    /// Fills the content of a preallocated MXC URI with the given content type
    /// and data.
    ///
    /// The URI must have been preallocated with [`Self::create_content_uri`].
    /// See this method's documentation for a full example.
    pub async fn upload_preallocated(
        &self,
        uri: PreallocatedMxcUri,
        content_type: &Mime,
        data: Vec<u8>,
    ) -> Result<()> {
        // Do a best-effort at reporting an expired MXC URI here; otherwise the server
        // may complain about it later.
        if let Some(expire_date) = uri.expire_date
            && MilliSecondsSinceUnixEpoch::now() >= expire_date
        {
            return Err(Error::Media(MediaError::ExpiredPreallocatedMxcUri));
        }

        let timeout = std::cmp::max(
            Duration::from_secs(data.len() as u64 / DEFAULT_UPLOAD_SPEED),
            MIN_UPLOAD_REQUEST_TIMEOUT,
        );

        let request = assign!(media::create_content_async::v3::Request::from_url(&uri.uri, data)?, {
            content_type: Some(content_type.as_ref().to_owned()),
        });

        let request_config = self.client.request_config().timeout(timeout);

        if let Err(err) = self.client.send(request).with_request_config(request_config).await {
            match err.client_api_error_kind() {
                Some(ErrorKind::CannotOverwriteMedia) => {
                    Err(Error::Media(MediaError::CannotOverwriteMedia))
                }

                // Unfortunately, the spec says a server will return 404 for either an expired MXC
                // ID or a non-existing MXC ID. Do a best-effort guess to recognize an expired MXC
                // ID based on the error string, which will work with Synapse (as of 2024-10-23).
                Some(ErrorKind::Unknown) if err.to_string().contains("expired") => {
                    Err(Error::Media(MediaError::ExpiredPreallocatedMxcUri))
                }

                _ => Err(err.into()),
            }
        } else {
            Ok(())
        }
    }

    /// Gets a media file by copying it to a temporary location on disk.
    ///
    /// The file won't be encrypted even if it is encrypted on the server.
    ///
    /// Returns a `MediaFileHandle` which takes ownership of the file. When the
    /// handle is dropped, the file will be deleted from the temporary location.
    ///
    /// # Arguments
    ///
    /// * `request` - The `MediaRequest` of the content.
    ///
    /// * `filename` - The filename specified in the event. It is suggested to
    ///   use the `filename()` method on the event's content instead of using
    ///   the `filename` field directly. If not provided, a random name will be
    ///   generated.
    ///
    /// * `content_type` - The type of the media, this will be used to set the
    ///   temporary file's extension when one isn't included in the filename.
    ///
    /// * `use_cache` - If we should use the media cache for this request.
    ///
    /// * `temp_dir` - Path to a directory where temporary directories can be
    ///   created. If not provided, a default, global temporary directory will
    ///   be used; this may not work properly on Android, where the default
    ///   location may require root access on some older Android versions.
    #[cfg(not(target_family = "wasm"))]
    pub async fn get_media_file(
        &self,
        request: &MediaRequestParameters,
        filename: Option<String>,
        content_type: &Mime,
        use_cache: bool,
        temp_dir: Option<String>,
    ) -> Result<MediaFileHandle> {
        let data = self.get_media_content(request, use_cache).await?;

        let inferred_extension = mime2ext::mime2ext(content_type);

        let filename_as_path = filename.as_ref().map(Path::new);

        let (sanitized_filename, filename_has_extension) = if let Some(path) = filename_as_path {
            let sanitized_filename = path.file_name().and_then(|f| f.to_str());
            let filename_has_extension = path.extension().is_some();
            (sanitized_filename, filename_has_extension)
        } else {
            (None, false)
        };

        let (temp_file, temp_dir) =
            match (sanitized_filename, filename_has_extension, inferred_extension) {
                // If the file name has an extension use that
                (Some(filename_with_extension), true, _) => {
                    // Use an intermediary directory to avoid conflicts
                    let temp_dir = temp_dir.map(TempDir::new_in).unwrap_or_else(TempDir::new)?;
                    let temp_file = TempFileBuilder::new()
                        .prefix(filename_with_extension)
                        .rand_bytes(0)
                        .tempfile_in(&temp_dir)?;
                    (temp_file, Some(temp_dir))
                }
                // If the file name doesn't have an extension try inferring one for it
                (Some(filename), false, Some(inferred_extension)) => {
                    // Use an intermediary directory to avoid conflicts
                    let temp_dir = temp_dir.map(TempDir::new_in).unwrap_or_else(TempDir::new)?;
                    let temp_file = TempFileBuilder::new()
                        .prefix(filename)
                        .suffix(&(".".to_owned() + inferred_extension))
                        .rand_bytes(0)
                        .tempfile_in(&temp_dir)?;
                    (temp_file, Some(temp_dir))
                }
                // If the only thing we have is an inferred extension then use that together with a
                // randomly generated file name
                (None, _, Some(inferred_extension)) => (
                    TempFileBuilder::new()
                        .suffix(&&(".".to_owned() + inferred_extension))
                        .tempfile()?,
                    None,
                ),
                // Otherwise just use a completely random file name
                _ => (TempFileBuilder::new().tempfile()?, None),
            };

        let mut file = TokioFile::from_std(temp_file.reopen()?);
        file.write_all(&data).await?;
        // Make sure the file metadata is flushed to disk.
        file.sync_all().await?;

        Ok(MediaFileHandle { file: temp_file, _directory: temp_dir })
    }

    /// Get a media file's content.
    ///
    /// If the content is encrypted and encryption is enabled, the content will
    /// be decrypted.
    ///
    /// # Arguments
    ///
    /// * `request` - The `MediaRequest` of the content.
    ///
    /// * `use_cache` - If we should use the media cache for this request.
    pub async fn get_media_content(
        &self,
        request: &MediaRequestParameters,
        use_cache: bool,
    ) -> Result<Vec<u8>> {
        // This is a local media. Force to read the media's content from the store: it
        // cannot exist somewhere else!
        if Self::is_local_uri(&request.source) {
            // Local medias are always cached with `MediaFormat::File`, be it the file
            // itself or its thumbnail (see `RoomSendQueue::cache_media`), so ignore the
            // requested format.
            let request = &MediaRequestParameters {
                source: request.source.clone(),
                format: MediaFormat::File,
            };

            if let Some(content) =
                self.client.media_store().lock().await?.get_media_content(request).await?
            {
                return Ok(content);
            } else {
                return Err(Error::MediaStore(Box::new(store::MediaStoreError::InvalidData {
                    details: format!("Media does not exist: `{request:?}`"),
                })));
            }
        }

        // Read from the cache: if it doesn't exist, the execution continues by reading
        // the media from the network.
        if use_cache
            && let Some(content) =
                self.client.media_store().lock().await?.get_media_content(request).await?
        {
            return Ok(content);
        }

        let content = self
            .client
            .inner
            .media_fetcher
            .read()
            .await
            .fetch_media_content(&self.client, request)
            .await?;

        if use_cache {
            self.client
                .media_store()
                .lock()
                .await?
                .add_media_content(request, content.clone(), IgnoreMediaRetentionPolicy::No)
                .await?;
        }

        Ok(content)
    }

    /// Remove a media file's content from the store.
    ///
    /// # Arguments
    ///
    /// * `request` - The `MediaRequest` of the content.
    pub async fn remove_media_content(&self, request: &MediaRequestParameters) -> Result<()> {
        Ok(self.client.media_store().lock().await?.remove_media_content(request).await?)
    }

    /// Delete all the media content corresponding to the given
    /// uri from the store.
    ///
    /// # Arguments
    ///
    /// * `uri` - The `MxcUri` of the files.
    pub async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<()> {
        Ok(self.client.media_store().lock().await?.remove_media_content_for_uri(uri).await?)
    }

    /// Get the file of the given media event content.
    ///
    /// If the content is encrypted and encryption is enabled, the content will
    /// be decrypted.
    ///
    /// Returns `Ok(None)` if the event content has no file.
    ///
    /// This is a convenience method that calls the
    /// [`get_media_content`](#method.get_media_content) method.
    ///
    /// # Arguments
    ///
    /// * `event_content` - The media event content.
    ///
    /// * `use_cache` - If we should use the media cache for this file.
    pub async fn get_file(
        &self,
        event_content: &impl MediaEventContent,
        use_cache: bool,
    ) -> Result<Option<Vec<u8>>> {
        let Some(source) = event_content.source() else { return Ok(None) };
        let file = self
            .get_media_content(
                &MediaRequestParameters { source, format: MediaFormat::File },
                use_cache,
            )
            .await?;
        Ok(Some(file))
    }

    /// Remove the file of the given media event content from the cache.
    ///
    /// This is a convenience method that calls the
    /// [`remove_media_content`](#method.remove_media_content) method.
    ///
    /// # Arguments
    ///
    /// * `event_content` - The media event content.
    pub async fn remove_file(&self, event_content: &impl MediaEventContent) -> Result<()> {
        if let Some(source) = event_content.source() {
            self.remove_media_content(&MediaRequestParameters {
                source,
                format: MediaFormat::File,
            })
            .await?;
        }

        Ok(())
    }

    /// Get a thumbnail of the given media event content.
    ///
    /// If the content is encrypted and encryption is enabled, the content will
    /// be decrypted.
    ///
    /// Returns `Ok(None)` if the event content has no thumbnail.
    ///
    /// This is a convenience method that calls the
    /// [`get_media_content`](#method.get_media_content) method.
    ///
    /// # Arguments
    ///
    /// * `event_content` - The media event content.
    ///
    /// * `settings` - The _desired_ settings of the thumbnail. The actual
    ///   thumbnail may not match the settings specified.
    ///
    /// * `use_cache` - If we should use the media cache for this thumbnail.
    pub async fn get_thumbnail(
        &self,
        event_content: &impl MediaEventContent,
        settings: MediaThumbnailSettings,
        use_cache: bool,
    ) -> Result<Option<Vec<u8>>> {
        let Some(source) = event_content.thumbnail_source() else { return Ok(None) };
        let thumbnail = self
            .get_media_content(
                &MediaRequestParameters { source, format: MediaFormat::Thumbnail(settings) },
                use_cache,
            )
            .await?;
        Ok(Some(thumbnail))
    }

    /// Remove the thumbnail of the given media event content from the cache.
    ///
    /// This is a convenience method that calls the
    /// [`remove_media_content`](#method.remove_media_content) method.
    ///
    /// # Arguments
    ///
    /// * `event_content` - The media event content.
    ///
    /// * `size` - The _desired_ settings of the thumbnail. Must match the
    ///   settings requested with [`get_thumbnail`](#method.get_thumbnail).
    pub async fn remove_thumbnail(
        &self,
        event_content: &impl MediaEventContent,
        settings: MediaThumbnailSettings,
    ) -> Result<()> {
        if let Some(source) = event_content.source() {
            self.remove_media_content(&MediaRequestParameters {
                source,
                format: MediaFormat::Thumbnail(settings),
            })
            .await?
        }

        Ok(())
    }

    /// Get a preview for a URL, as OpenGraph-like data.
    ///
    /// This is generated by the homeserver, which fetches the URL itself. Note
    /// that servers may disable this endpoint entirely, in which case this
    /// returns an error, and that using it in an encrypted room discloses the
    /// URL to the homeserver.
    ///
    /// Uses the authenticated endpoint when the homeserver supports it,
    /// falling back to the deprecated unauthenticated one otherwise.
    ///
    /// # Arguments
    ///
    /// * `url` - The URL to get a preview of.
    ///
    /// * `ts` - The preferred point in time to return a preview for, if the
    ///   homeserver supports returning previews for a given point in time.
    ///
    /// # Returns
    ///
    /// The OpenGraph-like data for the URL, if the homeserver returned any. It
    /// is returned as raw JSON, since the fields are not a fixed set: they
    /// mirror OpenGraph, with the addition of `matrix:image:size` for the
    /// image size in bytes, and `og:image` holding an MXC URI rather than an
    /// HTTP URL.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use matrix_sdk::Client;
    /// # use url::Url;
    /// # async {
    /// # let homeserver = Url::parse("http://localhost:8080")?;
    /// # let client = Client::new(homeserver).await?;
    /// if let Some(data) =
    ///     client.media().get_media_preview("https://matrix.org", None).await?
    /// {
    ///     println!("Preview data: {}", data.get());
    /// }
    /// # anyhow::Ok(()) };
    /// ```
    pub async fn get_media_preview(
        &self,
        url: &str,
        ts: Option<MilliSecondsSinceUnixEpoch>,
    ) -> Result<Option<Box<RawJsonValue>>> {
        // Use the authenticated endpoint when the server supports it.
        let supported_versions = self.client.supported_versions().await?;

        let use_auth = authenticated_media::get_media_preview::v1::Request::PATH_BUILDER
            .is_supported(&supported_versions);

        if use_auth {
            let mut request =
                authenticated_media::get_media_preview::v1::Request::new(url.to_owned());
            request.ts = ts;

            Ok(self.client.send(request).await?.data)
        } else {
            // The whole block is `allow(deprecated)`: both the endpoint and its
            // `ts` field are deprecated since Matrix 1.11, and we only reach
            // this branch when the homeserver is too old for the authenticated
            // endpoint above.
            #[allow(deprecated)]
            {
                let mut request = media::get_media_preview::v3::Request::new(url.to_owned());
                request.ts = ts;

                Ok(self.client.send(request).await?.data)
            }
        }
    }

    /// Set the [`MediaRetentionPolicy`] to use for deciding whether to store or
    /// keep media content.
    ///
    /// It is used:
    ///
    /// * When a media needs to be cached, to check that it does not exceed the
    ///   max file size.
    ///
    /// * When [`Media::clean()`], to check that all media content in the store
    ///   fits those criteria.
    ///
    /// To apply the new policy to the media cache right away,
    /// [`Media::clean()`] should be called after this.
    ///
    /// By default, an empty `MediaRetentionPolicy` is used, which means that no
    /// criteria are applied.
    ///
    /// # Arguments
    ///
    /// * `policy` - The `MediaRetentionPolicy` to use.
    pub async fn set_media_retention_policy(&self, policy: MediaRetentionPolicy) -> Result<()> {
        self.client.media_store().lock().await?.set_media_retention_policy(policy).await?;
        Ok(())
    }

    /// Get the current `MediaRetentionPolicy`.
    pub async fn media_retention_policy(&self) -> Result<MediaRetentionPolicy> {
        Ok(self.client.media_store().lock().await?.media_retention_policy())
    }

    /// Clean up the media cache with the current [`MediaRetentionPolicy`].
    ///
    /// If there is already an ongoing cleanup, this is a noop.
    pub async fn clean(&self) -> Result<()> {
        self.client.media_store().lock().await?.clean().await?;
        Ok(())
    }

    /// Upload the file bytes in `data` and return the source information.
    pub(crate) async fn upload_plain_media_and_thumbnail(
        &self,
        content_type: &Mime,
        data: Vec<u8>,
        thumbnail: Option<Thumbnail>,
        send_progress: SharedObservable<TransmissionProgress>,
    ) -> Result<(MediaSource, Option<(MediaSource, Box<ThumbnailInfo>)>)> {
        let upload_thumbnail = self.upload_thumbnail(thumbnail, send_progress.clone());

        let upload_attachment = async move {
            self.upload(content_type, data, None).with_send_progress_observable(send_progress).await
        };

        let (thumbnail, response) = try_join(upload_thumbnail, upload_attachment).await?;

        Ok((MediaSource::Plain(response.content_uri), thumbnail))
    }

    /// Uploads an unencrypted thumbnail to the media repository, and returns
    /// its source and extra information.
    async fn upload_thumbnail(
        &self,
        thumbnail: Option<Thumbnail>,
        send_progress: SharedObservable<TransmissionProgress>,
    ) -> Result<Option<(MediaSource, Box<ThumbnailInfo>)>> {
        let Some(thumbnail) = thumbnail else {
            return Ok(None);
        };

        let (data, content_type, thumbnail_info) = thumbnail.into_parts();

        let response = self
            .upload(&content_type, data, None)
            .with_send_progress_observable(send_progress)
            .await?;
        let url = response.content_uri;

        Ok(Some((MediaSource::Plain(url), thumbnail_info)))
    }

    /// Create an [`OwnedMxcUri`] for a file or thumbnail we want to store
    /// locally before sending it.
    ///
    /// This uses a MXC ID that is only locally valid.
    pub(crate) fn make_local_uri(txn_id: &TransactionId) -> OwnedMxcUri {
        OwnedMxcUri::from(format!("mxc://{LOCAL_MXC_SERVER_NAME}/{txn_id}"))
    }

    /// Create a [`MediaRequest`] for a file we want to store locally before
    /// sending it.
    ///
    /// This uses a MXC ID that is only locally valid.
    pub(crate) fn make_local_file_media_request(txn_id: &TransactionId) -> MediaRequestParameters {
        MediaRequestParameters {
            source: MediaSource::Plain(Self::make_local_uri(txn_id)),
            format: MediaFormat::File,
        }
    }

    /// Checks whether the MXC represents a local URI.
    ///
    /// A local MXC URI is a URI that was generated with
    /// [`Self::make_local_uri`].
    fn is_local_uri(source: &MediaSource) -> bool {
        let uri = match source {
            MediaSource::Plain(uri) => uri,
            MediaSource::Encrypted(file) => &file.url,
        };

        uri.server_name().is_ok_and(|server_name| server_name == LOCAL_MXC_SERVER_NAME)
    }
}

/// A [`MediaFetcher`] that uses the default media/authenticated media endpoints
/// to fetch new media.
#[derive(Debug, Clone)]
pub struct DefaultMediaFetcher;

impl MediaFetcher for DefaultMediaFetcher {
    fn fetch_media_content<'a>(
        &'a self,
        client: &'a Client,
        request: &'a MediaRequestParameters,
    ) -> BoxFuture<'a, Result<Vec<u8>, Error>> {
        Box::pin(async move {
            let request_config = client
                .request_config()
                // Downloading a file should have no timeout as we don't know the network
                // connectivity available for the user or the file size
                .timeout(Some(Duration::MAX));

            // Use the authenticated endpoints when the server supports it.
            let supported_versions = client.supported_versions().await?;

            let use_auth = authenticated_media::get_content::v1::Request::PATH_BUILDER
                .is_supported(&supported_versions);

            match &request.source {
                MediaSource::Encrypted(file) => {
                    let content = if use_auth {
                        let request =
                            authenticated_media::get_content::v1::Request::from_uri(&file.url)?;
                        client.send(request).with_request_config(request_config).await?.file
                    } else {
                        #[allow(deprecated)]
                        let request = media::get_content::v3::Request::from_url(&file.url)?;
                        client.send(request).with_request_config(request_config).await?.file
                    };

                    #[cfg(feature = "e2e-encryption")]
                    let content = {
                        let content_len = content.len();
                        let mut cursor = std::io::Cursor::new(content);
                        let mut reader = matrix_sdk_base::crypto::AttachmentDecryptor::new(
                            &mut cursor,
                            file.as_ref().clone().into(),
                        )?;

                        // Encrypted size should be the same as the decrypted size,
                        // rounded up to a cipher block.
                        let mut decrypted = Vec::with_capacity(content_len);

                        reader.read_to_end(&mut decrypted)?;

                        decrypted
                    };

                    Ok(content)
                }

                MediaSource::Plain(uri) => {
                    if let MediaFormat::Thumbnail(settings) = &request.format {
                        if use_auth {
                            let mut request =
                                authenticated_media::get_content_thumbnail::v1::Request::from_uri(
                                    uri,
                                    settings.width,
                                    settings.height,
                                )?;
                            request.method = Some(settings.method.clone());
                            request.animated = Some(settings.animated);

                            Ok(client.send(request).with_request_config(request_config).await?.file)
                        } else {
                            #[allow(deprecated)]
                            let request = {
                                let mut request =
                                    media::get_content_thumbnail::v3::Request::from_url(
                                        uri,
                                        settings.width,
                                        settings.height,
                                    )?;
                                request.method = Some(settings.method.clone());
                                request.animated = Some(settings.animated);
                                request
                            };

                            Ok(client.send(request).with_request_config(request_config).await?.file)
                        }
                    } else if use_auth {
                        let request = authenticated_media::get_content::v1::Request::from_uri(uri)?;
                        Ok(client.send(request).with_request_config(request_config).await?.file)
                    } else {
                        #[allow(deprecated)]
                        let request = media::get_content::v3::Request::from_url(uri)?;
                        Ok(client.send(request).with_request_config(request_config).await?.file)
                    }
                }
            }
        })
    }
}

#[cfg(test)]
mod tests {
    use std::ops::Not;

    use ruma::{
        MxcUri,
        events::room::{EncryptedFile, MediaSource},
        mxc_uri, owned_mxc_uri,
    };
    use serde_json::json;

    use super::Media;

    /// Create an `EncryptedFile` with the given MXC URI.
    fn encrypted_file(mxc_uri: &MxcUri) -> Box<EncryptedFile> {
        Box::new(
            serde_json::from_value(json!({
                "url": mxc_uri,
                "key": {
                    "kty": "oct",
                    "key_ops": ["encrypt", "decrypt"],
                    "alg": "A256CTR",
                    "k": "b50ACIv6LMn9AfMCFD1POJI_UAFWIclxAN1kWrEO2X8",
                    "ext": true,
                },
                "iv": "AK1wyzigZtQAAAABAAAAKK",
                "hashes": {
                    "sha256": "/NogKqW5bz/m8xHgFiH5haFGjCNVmUIPLzfvOhHdrxY",
                },
                "v": "v2",
            }))
            .unwrap(),
        )
    }

    #[test]
    fn test_make_local_uri() {
        let txn_id = "abcdef";

        let uri = Media::make_local_uri(txn_id.into());
        assert_eq!(uri.media_id().unwrap(), txn_id);
    }

    #[test]
    fn test_is_local_uri() {
        let txn_id = "abcdef";

        // Request generated with `make_local_file_media_request`.
        let request = Media::make_local_file_media_request(txn_id.into());
        assert!(Media::is_local_uri(&request.source));

        // Local plain source.
        let source = MediaSource::Plain(Media::make_local_uri(txn_id.into()));
        assert!(Media::is_local_uri(&source));

        // Local encrypted source.
        let source = MediaSource::Encrypted(encrypted_file(&Media::make_local_uri(txn_id.into())));
        assert!(Media::is_local_uri(&source));

        // Test non-local plain source.
        let source = MediaSource::Plain(owned_mxc_uri!("mxc://server.local/poiuyt"));
        assert!(Media::is_local_uri(&source).not());

        // Test non-local encrypted source.
        let source = MediaSource::Encrypted(encrypted_file(mxc_uri!("mxc://server.local/mlkjhg")));
        assert!(Media::is_local_uri(&source).not());

        // Test invalid MXC URI.
        let source = MediaSource::Plain("https://server.local/nbvcxw".into());
        assert!(Media::is_local_uri(&source).not());
    }
}