From b562ceb92aba71c55f655ba0b2a97da24766f7a4 Mon Sep 17 00:00:00 2001 From: Shrey Patel Date: Fri, 22 Aug 2025 11:40:51 +0100 Subject: [PATCH] Instantiate `MediaStore`s and use where required. --- bindings/matrix-sdk-ffi/src/client.rs | 13 +- crates/matrix-sdk-base/src/client.rs | 11 + .../src/media/store/integration_tests.rs | 11 +- .../src/media/store/media_service.rs | 279 +----------- crates/matrix-sdk-base/src/media/store/mod.rs | 104 ++++- .../matrix-sdk-base/src/media/store/traits.rs | 427 ++++++++++++++++++ crates/matrix-sdk-base/src/store/mod.rs | 18 + .../src/event_cache_store/mod.rs | 5 +- crates/matrix-sdk/src/client/builder/mod.rs | 11 +- crates/matrix-sdk/src/client/mod.rs | 6 + crates/matrix-sdk/src/error.rs | 14 +- crates/matrix-sdk/src/lib.rs | 3 +- crates/matrix-sdk/src/media.rs | 20 +- crates/matrix-sdk/src/room/mod.rs | 9 +- crates/matrix-sdk/src/send_queue/mod.rs | 8 +- crates/matrix-sdk/src/send_queue/progress.rs | 10 +- crates/matrix-sdk/src/send_queue/upload.rs | 54 +-- 17 files changed, 657 insertions(+), 346 deletions(-) create mode 100644 crates/matrix-sdk-base/src/media/store/traits.rs diff --git a/bindings/matrix-sdk-ffi/src/client.rs b/bindings/matrix-sdk-ffi/src/client.rs index e4c7306bf07..c62e32c0ed9 100644 --- a/bindings/matrix-sdk-ffi/src/client.rs +++ b/bindings/matrix-sdk-ffi/src/client.rs @@ -14,7 +14,6 @@ use matrix_sdk::{ authentication::oauth::{ AccountManagementActionFull, ClientId, OAuthAuthorizationData, OAuthSession, }, - event_cache::EventCacheError, media::{MediaFormat, MediaRequestParameters, MediaRetentionPolicy, MediaThumbnailSettings}, ruma::{ api::client::{ @@ -39,7 +38,7 @@ use matrix_sdk::{ }, sliding_sync::Version as SdkSlidingSyncVersion, store::RoomLoadSettings as SdkRoomLoadSettings, - Account, AuthApi, AuthSession, Client as MatrixClient, SessionChange, SessionTokens, + Account, AuthApi, AuthSession, Client as MatrixClient, Error, SessionChange, SessionTokens, STATE_STORE_DATABASE_NAME, }; use matrix_sdk_common::{stream::StreamExt, SendOutsideWasm, SyncOutsideWasm}; @@ -1510,8 +1509,8 @@ impl Client { &self, policy: MediaRetentionPolicy, ) -> Result<(), ClientError> { - let closure = async || -> Result<_, EventCacheError> { - let store = self.inner.event_cache_store().lock().await?; + let closure = async || -> Result<_, Error> { + let store = self.inner.media_store().lock().await?; Ok(store.set_media_retention_policy(policy).await?) }; @@ -1559,13 +1558,13 @@ impl Client { // Clean up the media cache according to the current media retention policy. self.inner - .event_cache_store() + .media_store() .lock() .await - .map_err(EventCacheError::from)? + .map_err(Error::from)? .clean_up_media_cache() .await - .map_err(EventCacheError::from)?; + .map_err(Error::from)?; // Clear all the room chunks. It's important to *not* call // `EventCacheStore::clear_all_linked_chunks` here, because there might be live diff --git a/crates/matrix-sdk-base/src/client.rs b/crates/matrix-sdk-base/src/client.rs index 250ed4e589e..4318e981d27 100644 --- a/crates/matrix-sdk-base/src/client.rs +++ b/crates/matrix-sdk-base/src/client.rs @@ -57,6 +57,7 @@ use crate::{ deserialized_responses::DisplayName, error::{Error, Result}, event_cache::store::EventCacheStoreLock, + media::store::MediaStoreLock, response_processors::{self as processors, Context}, room::{ Room, RoomInfoNotableUpdate, RoomInfoNotableUpdateReasons, RoomMembersUpdate, RoomState, @@ -91,6 +92,9 @@ pub struct BaseClient { /// The store used by the event cache. event_cache_store: EventCacheStoreLock, + /// The store used by the media cache. + media_store: MediaStoreLock, + /// The store used for encryption. /// /// This field is only meant to be used for `OlmMachine` initialization. @@ -189,6 +193,7 @@ impl BaseClient { BaseClient { state_store: store, event_cache_store: config.event_cache_store, + media_store: config.media_store, #[cfg(feature = "e2e-encryption")] crypto_store: config.crypto_store, #[cfg(feature = "e2e-encryption")] @@ -222,6 +227,7 @@ impl BaseClient { let copy = Self { state_store: BaseStateStore::new(config.state_store), event_cache_store: config.event_cache_store, + media_store: config.media_store, // We copy the crypto store as well as the `OlmMachine` for two reasons: // 1. The `self.crypto_store` is the same as the one used inside the `OlmMachine`. // 2. We need to ensure that the parent and child use the same data and caches inside @@ -306,6 +312,11 @@ impl BaseClient { &self.event_cache_store } + /// Get a reference to the media store. + pub fn media_store(&self) -> &MediaStoreLock { + &self.media_store + } + /// Check whether the client has been activated. /// /// See [`BaseClient::activate`] to know what it means. diff --git a/crates/matrix-sdk-base/src/media/store/integration_tests.rs b/crates/matrix-sdk-base/src/media/store/integration_tests.rs index 9400bcbc24e..1df1d805653 100644 --- a/crates/matrix-sdk-base/src/media/store/integration_tests.rs +++ b/crates/matrix-sdk-base/src/media/store/integration_tests.rs @@ -23,11 +23,10 @@ use ruma::{ uint, }; -use super::{ - MediaRetentionPolicy, MediaStoreInner, - media_service::{IgnoreMediaRetentionPolicy, MediaStore}, +use super::{MediaRetentionPolicy, MediaStoreInner, media_service::IgnoreMediaRetentionPolicy}; +use crate::media::{ + MediaFormat, MediaRequestParameters, MediaThumbnailSettings, store::MediaStore, }; -use crate::media::{MediaFormat, MediaRequestParameters, MediaThumbnailSettings}; /// [`MediaStoreInner`] integration tests. /// @@ -981,7 +980,7 @@ where /// ```no_run /// # use matrix_sdk_base::media::store::{ /// # MediaStore, -/// # MemoryStore as MyStore, +/// # MemoryMediaStore as MyStore, /// # Result as MediaStoreResult, /// # }; /// @@ -1275,7 +1274,7 @@ where /// ```no_run /// # use matrix_sdk_base::media::store::{ /// # MediaStore, -/// # MemoryStore as MyStore, +/// # MemoryMediaStore as MyStore, /// # Result as MediaStoreResult, /// # }; /// diff --git a/crates/matrix-sdk-base/src/media/store/media_service.rs b/crates/matrix-sdk-base/src/media/store/media_service.rs index 5250634124e..7c787c509fc 100644 --- a/crates/matrix-sdk-base/src/media/store/media_service.rs +++ b/crates/matrix-sdk-base/src/media/store/media_service.rs @@ -12,11 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{fmt, sync::Arc}; +use std::sync::Arc; -use async_trait::async_trait; use matrix_sdk_common::{ - AsyncTraitDeps, SendOutsideWasm, SyncOutsideWasm, + SendOutsideWasm, SyncOutsideWasm, executor::{JoinHandle, spawn}, locks::Mutex, }; @@ -24,7 +23,7 @@ use ruma::{MxcUri, time::SystemTime}; use tokio::sync::Mutex as AsyncMutex; use tracing::error; -use super::{MediaRetentionPolicy, MediaStoreError}; +use super::{MediaRetentionPolicy, MediaStoreInner}; use crate::media::MediaRequestParameters; /// API for implementors of [`MediaStore`] to manage their media through @@ -349,272 +348,6 @@ where } } -/// An abstract trait that can be used to implement different store backends -/// for the media store of the SDK. -#[cfg_attr(target_family = "wasm", async_trait(?Send))] -#[cfg_attr(not(target_family = "wasm"), async_trait)] -pub trait MediaStore: AsyncTraitDeps { - /// The error type used by this media store. - type Error: fmt::Debug + Into; - - /// Try to take a lock using the given store. - async fn try_take_leased_lock( - &self, - lease_duration_ms: u32, - key: &str, - holder: &str, - ) -> Result; - - /// Add a media file's content in the media store. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequest` of the file. - /// - /// * `content` - The content of the file. - async fn add_media_content( - &self, - request: &MediaRequestParameters, - content: Vec, - ignore_policy: IgnoreMediaRetentionPolicy, - ) -> Result<(), Self::Error>; - - /// Replaces the given media's content key with another one. - /// - /// This should be used whenever a temporary (local) MXID has been used, and - /// it must now be replaced with its actual remote counterpart (after - /// uploading some content, or creating an empty MXC URI). - /// - /// ⚠ No check is performed to ensure that the media formats are consistent, - /// i.e. it's possible to update with a thumbnail key a media that was - /// keyed as a file before. The caller is responsible of ensuring that - /// the replacement makes sense, according to their use case. - /// - /// This should not raise an error when the `from` parameter points to an - /// unknown media, and it should silently continue in this case. - /// - /// # Arguments - /// - /// * `from` - The previous `MediaRequest` of the file. - /// - /// * `to` - The new `MediaRequest` of the file. - async fn replace_media_key( - &self, - from: &MediaRequestParameters, - to: &MediaRequestParameters, - ) -> Result<(), Self::Error>; - - /// Get a media file's content out of the media store. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequest` of the file. - async fn get_media_content( - &self, - request: &MediaRequestParameters, - ) -> Result>, Self::Error>; - - /// Remove a media file's content from the media store. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequest` of the file. - async fn remove_media_content( - &self, - request: &MediaRequestParameters, - ) -> Result<(), Self::Error>; - - /// Get a media file's content associated to an `MxcUri` from the - /// media store. - /// - /// In theory, there could be several files stored using the same URI and a - /// different `MediaFormat`. This API is meant to be used with a media file - /// that has only been stored with a single format. - /// - /// If there are several media files for a given URI in different formats, - /// this API will only return one of them. Which one is left as an - /// implementation detail. - /// - /// # Arguments - /// - /// * `uri` - The `MxcUri` of the media file. - async fn get_media_content_for_uri(&self, uri: &MxcUri) - -> Result>, Self::Error>; - - /// Remove all the media files' content associated to an `MxcUri` from the - /// media store. - /// - /// This should not raise an error when the `uri` parameter points to an - /// unknown media, and it should return an Ok result in this case. - /// - /// # Arguments - /// - /// * `uri` - The `MxcUri` of the media files. - async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<(), Self::Error>; - - /// Set the `MediaRetentionPolicy` to use for deciding whether to store or - /// keep media content. - /// - /// # Arguments - /// - /// * `policy` - The `MediaRetentionPolicy` to use. - async fn set_media_retention_policy( - &self, - policy: MediaRetentionPolicy, - ) -> Result<(), Self::Error>; - - /// Get the current `MediaRetentionPolicy`. - fn media_retention_policy(&self) -> MediaRetentionPolicy; - - /// Set whether the current [`MediaRetentionPolicy`] should be ignored for - /// the media. - /// - /// The change will be taken into account in the next cleanup. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequestParameters` of the file. - /// - /// * `ignore_policy` - Whether the current `MediaRetentionPolicy` should be - /// ignored. - async fn set_ignore_media_retention_policy( - &self, - request: &MediaRequestParameters, - ignore_policy: IgnoreMediaRetentionPolicy, - ) -> Result<(), Self::Error>; - - /// Clean up the media cache with the current `MediaRetentionPolicy`. - /// - /// If there is already an ongoing cleanup, this is a noop. - async fn clean_up_media_cache(&self) -> Result<(), Self::Error>; -} - -/// An abstract trait that can be used to implement different store backends -/// for the media cache of the SDK. -/// -/// The main purposes of this trait are to be able to centralize where we handle -/// [`MediaRetentionPolicy`] by wrapping this in a [`MediaService`], and to -/// simplify the implementation of tests by being able to have complete control -/// over the `SystemTime`s provided to the store. -#[cfg_attr(target_family = "wasm", async_trait(?Send))] -#[cfg_attr(not(target_family = "wasm"), async_trait)] -pub trait MediaStoreInner: AsyncTraitDeps + Clone { - /// The error type used by this media cache store. - type Error: fmt::Debug + fmt::Display + Into; - - /// The persisted media retention policy in the media cache. - async fn media_retention_policy_inner( - &self, - ) -> Result, Self::Error>; - - /// Persist the media retention policy in the media cache. - /// - /// # Arguments - /// - /// * `policy` - The `MediaRetentionPolicy` to persist. - async fn set_media_retention_policy_inner( - &self, - policy: MediaRetentionPolicy, - ) -> Result<(), Self::Error>; - - /// Add a media file's content in the media cache. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequestParameters` of the file. - /// - /// * `content` - The content of the file. - /// - /// * `current_time` - The current time, to set the last access time of the - /// media. - /// - /// * `policy` - The media retention policy, to check whether the media is - /// too big to be cached. - /// - /// * `ignore_policy` - Whether the `MediaRetentionPolicy` should be ignored - /// for this media. This setting should be persisted alongside the media - /// and taken into account whenever the policy is used. - async fn add_media_content_inner( - &self, - request: &MediaRequestParameters, - content: Vec, - current_time: SystemTime, - policy: MediaRetentionPolicy, - ignore_policy: IgnoreMediaRetentionPolicy, - ) -> Result<(), Self::Error>; - - /// Set whether the current [`MediaRetentionPolicy`] should be ignored for - /// the media. - /// - /// If the media of the given request is not found, this should be a noop. - /// - /// The change will be taken into account in the next cleanup. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequestParameters` of the file. - /// - /// * `ignore_policy` - Whether the current `MediaRetentionPolicy` should be - /// ignored. - async fn set_ignore_media_retention_policy_inner( - &self, - request: &MediaRequestParameters, - ignore_policy: IgnoreMediaRetentionPolicy, - ) -> Result<(), Self::Error>; - - /// Get a media file's content out of the media cache. - /// - /// # Arguments - /// - /// * `request` - The `MediaRequestParameters` of the file. - /// - /// * `current_time` - The current time, to update the last access time of - /// the media. - async fn get_media_content_inner( - &self, - request: &MediaRequestParameters, - current_time: SystemTime, - ) -> Result>, Self::Error>; - - /// Get a media file's content associated to an `MxcUri` from the - /// media store. - /// - /// # Arguments - /// - /// * `uri` - The `MxcUri` of the media file. - /// - /// * `current_time` - The current time, to update the last access time of - /// the media. - async fn get_media_content_for_uri_inner( - &self, - uri: &MxcUri, - current_time: SystemTime, - ) -> Result>, Self::Error>; - - /// Clean up the media cache with the given policy. - /// - /// For the integration tests, it is expected that content that does not - /// pass the last access expiry and max file size criteria will be - /// removed first. After that, the remaining cache size should be - /// computed to compare against the max cache size criteria. - /// - /// # Arguments - /// - /// * `policy` - The media retention policy to use for the cleanup. The - /// `cleanup_frequency` will be ignored. - /// - /// * `current_time` - The current time, to be used to check for expired - /// content and to be stored as the time of the last media cache cleanup. - async fn clean_up_media_cache_inner( - &self, - policy: MediaRetentionPolicy, - current_time: SystemTime, - ) -> Result<(), Self::Error>; - - /// The time of the last media cache cleanup. - async fn last_media_cleanup_time_inner(&self) -> Result, Self::Error>; -} - /// Whether the [`MediaRetentionPolicy`] should be ignored for the current /// content. /// @@ -685,10 +418,10 @@ mod tests { }; use super::{ - IgnoreMediaRetentionPolicy, MediaRetentionPolicy, MediaService, MediaStoreError, - MediaStoreInner, TimeProvider, + IgnoreMediaRetentionPolicy, MediaRetentionPolicy, MediaService, MediaStoreInner, + TimeProvider, }; - use crate::media::{MediaFormat, MediaRequestParameters, UniqueKey}; + use crate::media::{MediaFormat, MediaRequestParameters, UniqueKey, store::MediaStoreError}; #[derive(Debug, Default, Clone)] struct MockMediaStoreInner { diff --git a/crates/matrix-sdk-base/src/media/store/mod.rs b/crates/matrix-sdk-base/src/media/store/mod.rs index d49ea7031eb..f63628c3f7d 100644 --- a/crates/matrix-sdk-base/src/media/store/mod.rs +++ b/crates/matrix-sdk-base/src/media/store/mod.rs @@ -22,17 +22,26 @@ mod media_retention_policy; mod media_service; mod memory_store; +mod traits; #[cfg(any(test, feature = "testing"))] #[macro_use] pub mod integration_tests; +#[cfg(not(tarpaulin_include))] +use std::fmt; +use std::{ops::Deref, sync::Arc}; + +use matrix_sdk_common::store_locks::{ + BackingStore, CrossProcessStoreLock, CrossProcessStoreLockGuard, LockStoreError, +}; use matrix_sdk_store_encryption::Error as StoreEncryptionError; +pub use traits::{DynMediaStore, IntoMediaStore, MediaStore, MediaStoreInner}; #[cfg(any(test, feature = "testing"))] pub use self::integration_tests::{MediaStoreInnerIntegrationTests, MediaStoreIntegrationTests}; pub use self::{ media_retention_policy::MediaRetentionPolicy, - media_service::{IgnoreMediaRetentionPolicy, MediaService, MediaStore, MediaStoreInner}, + media_service::{IgnoreMediaRetentionPolicy, MediaService}, memory_store::MemoryMediaStore, }; @@ -74,3 +83,96 @@ impl MediaStoreError { /// An `MediaStore` specific result type. pub type Result = std::result::Result; + +/// The high-level public type to represent an `MediaStore` lock. +#[derive(Clone)] +pub struct MediaStoreLock { + /// The inner cross process lock that is used to lock the `MediaStore`. + cross_process_lock: Arc>, + + /// The store itself. + /// + /// That's the only place where the store exists. + store: Arc, +} + +#[cfg(not(tarpaulin_include))] +impl fmt::Debug for MediaStoreLock { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.debug_struct("MediaStoreLock").finish_non_exhaustive() + } +} + +impl MediaStoreLock { + /// Create a new lock around the [`MediaStore`]. + /// + /// The `holder` argument represents the holder inside the + /// [`CrossProcessStoreLock::new`]. + pub fn new(store: S, holder: String) -> Self + where + S: IntoMediaStore, + { + let store = store.into_event_cache_store(); + + Self { + cross_process_lock: Arc::new(CrossProcessStoreLock::new( + LockableMediaStore(store.clone()), + "default".to_owned(), + holder, + )), + store, + } + } + + /// Acquire a spin lock (see [`CrossProcessStoreLock::spin_lock`]). + pub async fn lock(&self) -> Result, LockStoreError> { + let cross_process_lock_guard = self.cross_process_lock.spin_lock(None).await?; + + Ok(MediaStoreLockGuard { cross_process_lock_guard, store: self.store.deref() }) + } +} + +/// An RAII implementation of a “scoped lock” of an [`MediaStoreLock`]. +/// When this structure is dropped (falls out of scope), the lock will be +/// unlocked. +pub struct MediaStoreLockGuard<'a> { + /// The cross process lock guard. + #[allow(unused)] + cross_process_lock_guard: CrossProcessStoreLockGuard, + + /// A reference to the store. + store: &'a DynMediaStore, +} + +#[cfg(not(tarpaulin_include))] +impl fmt::Debug for MediaStoreLockGuard<'_> { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.debug_struct("MediaStoreLockGuard").finish_non_exhaustive() + } +} + +impl Deref for MediaStoreLockGuard<'_> { + type Target = DynMediaStore; + + fn deref(&self) -> &Self::Target { + self.store + } +} + +/// A type that wraps the [`MediaStore`] but implements [`BackingStore`] to +/// make it usable inside the cross process lock. +#[derive(Clone, Debug)] +struct LockableMediaStore(Arc); + +impl BackingStore for LockableMediaStore { + type LockError = MediaStoreError; + + async fn try_lock( + &self, + lease_duration_ms: u32, + key: &str, + holder: &str, + ) -> std::result::Result { + self.0.try_take_leased_lock(lease_duration_ms, key, holder).await + } +} diff --git a/crates/matrix-sdk-base/src/media/store/traits.rs b/crates/matrix-sdk-base/src/media/store/traits.rs new file mode 100644 index 00000000000..8b287edcc22 --- /dev/null +++ b/crates/matrix-sdk-base/src/media/store/traits.rs @@ -0,0 +1,427 @@ +// Copyright 2025 Kévin Commaille +// +// 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. + +//! Types and traits regarding media caching of the event cache store. + +use std::{fmt, sync::Arc}; + +use async_trait::async_trait; +use matrix_sdk_common::AsyncTraitDeps; +use ruma::{MxcUri, time::SystemTime}; + +#[cfg(doc)] +use crate::media::store::MediaService; +use crate::media::{ + MediaRequestParameters, + store::{IgnoreMediaRetentionPolicy, MediaRetentionPolicy, MediaStoreError}, +}; + +/// An abstract trait that can be used to implement different store backends +/// for the event cache of the SDK. +#[cfg_attr(target_family = "wasm", async_trait(?Send))] +#[cfg_attr(not(target_family = "wasm"), async_trait)] +pub trait MediaStore: AsyncTraitDeps { + /// The error type used by this event cache store. + type Error: fmt::Debug + Into; + + /// Try to take a lock using the given store. + async fn try_take_leased_lock( + &self, + lease_duration_ms: u32, + key: &str, + holder: &str, + ) -> Result; + + /// Add a media file's content in the media store. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequest` of the file. + /// + /// * `content` - The content of the file. + async fn add_media_content( + &self, + request: &MediaRequestParameters, + content: Vec, + ignore_policy: IgnoreMediaRetentionPolicy, + ) -> Result<(), Self::Error>; + + /// Replaces the given media's content key with another one. + /// + /// This should be used whenever a temporary (local) MXID has been used, and + /// it must now be replaced with its actual remote counterpart (after + /// uploading some content, or creating an empty MXC URI). + /// + /// ⚠ No check is performed to ensure that the media formats are consistent, + /// i.e. it's possible to update with a thumbnail key a media that was + /// keyed as a file before. The caller is responsible of ensuring that + /// the replacement makes sense, according to their use case. + /// + /// This should not raise an error when the `from` parameter points to an + /// unknown media, and it should silently continue in this case. + /// + /// # Arguments + /// + /// * `from` - The previous `MediaRequest` of the file. + /// + /// * `to` - The new `MediaRequest` of the file. + async fn replace_media_key( + &self, + from: &MediaRequestParameters, + to: &MediaRequestParameters, + ) -> Result<(), Self::Error>; + + /// Get a media file's content out of the media store. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequest` of the file. + async fn get_media_content( + &self, + request: &MediaRequestParameters, + ) -> Result>, Self::Error>; + + /// Remove a media file's content from the media store. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequest` of the file. + async fn remove_media_content( + &self, + request: &MediaRequestParameters, + ) -> Result<(), Self::Error>; + + /// Get a media file's content associated to an `MxcUri` from the + /// media store. + /// + /// In theory, there could be several files stored using the same URI and a + /// different `MediaFormat`. This API is meant to be used with a media file + /// that has only been stored with a single format. + /// + /// If there are several media files for a given URI in different formats, + /// this API will only return one of them. Which one is left as an + /// implementation detail. + /// + /// # Arguments + /// + /// * `uri` - The `MxcUri` of the media file. + async fn get_media_content_for_uri(&self, uri: &MxcUri) + -> Result>, Self::Error>; + + /// Remove all the media files' content associated to an `MxcUri` from the + /// media store. + /// + /// This should not raise an error when the `uri` parameter points to an + /// unknown media, and it should return an Ok result in this case. + /// + /// # Arguments + /// + /// * `uri` - The `MxcUri` of the media files. + async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<(), Self::Error>; + + /// Set the `MediaRetentionPolicy` to use for deciding whether to store or + /// keep media content. + /// + /// # Arguments + /// + /// * `policy` - The `MediaRetentionPolicy` to use. + async fn set_media_retention_policy( + &self, + policy: MediaRetentionPolicy, + ) -> Result<(), Self::Error>; + + /// Get the current `MediaRetentionPolicy`. + fn media_retention_policy(&self) -> MediaRetentionPolicy; + + /// Set whether the current [`MediaRetentionPolicy`] should be ignored for + /// the media. + /// + /// The change will be taken into account in the next cleanup. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequestParameters` of the file. + /// + /// * `ignore_policy` - Whether the current `MediaRetentionPolicy` should be + /// ignored. + async fn set_ignore_media_retention_policy( + &self, + request: &MediaRequestParameters, + ignore_policy: IgnoreMediaRetentionPolicy, + ) -> Result<(), Self::Error>; + + /// Clean up the media cache with the current `MediaRetentionPolicy`. + /// + /// If there is already an ongoing cleanup, this is a noop. + async fn clean_up_media_cache(&self) -> Result<(), Self::Error>; +} + +/// An abstract trait that can be used to implement different store backends +/// for the media cache of the SDK. +/// +/// The main purposes of this trait are to be able to centralize where we handle +/// [`MediaRetentionPolicy`] by wrapping this in a [`MediaService`], and to +/// simplify the implementation of tests by being able to have complete control +/// over the `SystemTime`s provided to the store. +#[cfg_attr(target_family = "wasm", async_trait(?Send))] +#[cfg_attr(not(target_family = "wasm"), async_trait)] +pub trait MediaStoreInner: AsyncTraitDeps + Clone { + /// The error type used by this media cache store. + type Error: fmt::Debug + fmt::Display + Into; + + /// The persisted media retention policy in the media cache. + async fn media_retention_policy_inner( + &self, + ) -> Result, Self::Error>; + + /// Persist the media retention policy in the media cache. + /// + /// # Arguments + /// + /// * `policy` - The `MediaRetentionPolicy` to persist. + async fn set_media_retention_policy_inner( + &self, + policy: MediaRetentionPolicy, + ) -> Result<(), Self::Error>; + + /// Add a media file's content in the media cache. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequestParameters` of the file. + /// + /// * `content` - The content of the file. + /// + /// * `current_time` - The current time, to set the last access time of the + /// media. + /// + /// * `policy` - The media retention policy, to check whether the media is + /// too big to be cached. + /// + /// * `ignore_policy` - Whether the `MediaRetentionPolicy` should be ignored + /// for this media. This setting should be persisted alongside the media + /// and taken into account whenever the policy is used. + async fn add_media_content_inner( + &self, + request: &MediaRequestParameters, + content: Vec, + current_time: SystemTime, + policy: MediaRetentionPolicy, + ignore_policy: IgnoreMediaRetentionPolicy, + ) -> Result<(), Self::Error>; + + /// Set whether the current [`MediaRetentionPolicy`] should be ignored for + /// the media. + /// + /// If the media of the given request is not found, this should be a noop. + /// + /// The change will be taken into account in the next cleanup. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequestParameters` of the file. + /// + /// * `ignore_policy` - Whether the current `MediaRetentionPolicy` should be + /// ignored. + async fn set_ignore_media_retention_policy_inner( + &self, + request: &MediaRequestParameters, + ignore_policy: IgnoreMediaRetentionPolicy, + ) -> Result<(), Self::Error>; + + /// Get a media file's content out of the media cache. + /// + /// # Arguments + /// + /// * `request` - The `MediaRequestParameters` of the file. + /// + /// * `current_time` - The current time, to update the last access time of + /// the media. + async fn get_media_content_inner( + &self, + request: &MediaRequestParameters, + current_time: SystemTime, + ) -> Result>, Self::Error>; + + /// Get a media file's content associated to an `MxcUri` from the + /// media store. + /// + /// # Arguments + /// + /// * `uri` - The `MxcUri` of the media file. + /// + /// * `current_time` - The current time, to update the last access time of + /// the media. + async fn get_media_content_for_uri_inner( + &self, + uri: &MxcUri, + current_time: SystemTime, + ) -> Result>, Self::Error>; + + /// Clean up the media cache with the given policy. + /// + /// For the integration tests, it is expected that content that does not + /// pass the last access expiry and max file size criteria will be + /// removed first. After that, the remaining cache size should be + /// computed to compare against the max cache size criteria. + /// + /// # Arguments + /// + /// * `policy` - The media retention policy to use for the cleanup. The + /// `cleanup_frequency` will be ignored. + /// + /// * `current_time` - The current time, to be used to check for expired + /// content and to be stored as the time of the last media cache cleanup. + async fn clean_up_media_cache_inner( + &self, + policy: MediaRetentionPolicy, + current_time: SystemTime, + ) -> Result<(), Self::Error>; + + /// The time of the last media cache cleanup. + async fn last_media_cleanup_time_inner(&self) -> Result, Self::Error>; +} + +#[repr(transparent)] +struct EraseMediaStoreError(T); + +#[cfg(not(tarpaulin_include))] +impl fmt::Debug for EraseMediaStoreError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +#[cfg_attr(target_family = "wasm", async_trait(?Send))] +#[cfg_attr(not(target_family = "wasm"), async_trait)] +impl MediaStore for EraseMediaStoreError { + type Error = MediaStoreError; + + async fn try_take_leased_lock( + &self, + lease_duration_ms: u32, + key: &str, + holder: &str, + ) -> Result { + self.0.try_take_leased_lock(lease_duration_ms, key, holder).await.map_err(Into::into) + } + + async fn add_media_content( + &self, + request: &MediaRequestParameters, + content: Vec, + ignore_policy: IgnoreMediaRetentionPolicy, + ) -> Result<(), Self::Error> { + self.0.add_media_content(request, content, ignore_policy).await.map_err(Into::into) + } + + async fn replace_media_key( + &self, + from: &MediaRequestParameters, + to: &MediaRequestParameters, + ) -> Result<(), Self::Error> { + self.0.replace_media_key(from, to).await.map_err(Into::into) + } + + async fn get_media_content( + &self, + request: &MediaRequestParameters, + ) -> Result>, Self::Error> { + self.0.get_media_content(request).await.map_err(Into::into) + } + + async fn remove_media_content( + &self, + request: &MediaRequestParameters, + ) -> Result<(), Self::Error> { + self.0.remove_media_content(request).await.map_err(Into::into) + } + + async fn get_media_content_for_uri( + &self, + uri: &MxcUri, + ) -> Result>, Self::Error> { + self.0.get_media_content_for_uri(uri).await.map_err(Into::into) + } + + async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<(), Self::Error> { + self.0.remove_media_content_for_uri(uri).await.map_err(Into::into) + } + + async fn set_media_retention_policy( + &self, + policy: MediaRetentionPolicy, + ) -> Result<(), Self::Error> { + self.0.set_media_retention_policy(policy).await.map_err(Into::into) + } + + fn media_retention_policy(&self) -> MediaRetentionPolicy { + self.0.media_retention_policy() + } + + async fn set_ignore_media_retention_policy( + &self, + request: &MediaRequestParameters, + ignore_policy: IgnoreMediaRetentionPolicy, + ) -> Result<(), Self::Error> { + self.0.set_ignore_media_retention_policy(request, ignore_policy).await.map_err(Into::into) + } + + async fn clean_up_media_cache(&self) -> Result<(), Self::Error> { + self.0.clean_up_media_cache().await.map_err(Into::into) + } +} + +/// A type-erased [`MediaStore`]. +pub type DynMediaStore = dyn MediaStore; + +/// A type that can be type-erased into `Arc`. +/// +/// This trait is not meant to be implemented directly outside +/// `matrix-sdk-base`, but it is automatically implemented for everything that +/// implements `MediaStore`. +pub trait IntoMediaStore { + #[doc(hidden)] + fn into_event_cache_store(self) -> Arc; +} + +impl IntoMediaStore for Arc { + fn into_event_cache_store(self) -> Arc { + self + } +} + +impl IntoMediaStore for T +where + T: MediaStore + Sized + 'static, +{ + fn into_event_cache_store(self) -> Arc { + Arc::new(EraseMediaStoreError(self)) + } +} + +// Turns a given `Arc` into `Arc` by attaching the +// `MediaStore` impl vtable of `EraseMediaStoreError`. +impl IntoMediaStore for Arc +where + T: MediaStore + 'static, +{ + fn into_event_cache_store(self) -> Arc { + let ptr: *const T = Arc::into_raw(self); + let ptr_erased = ptr as *const EraseMediaStoreError; + // SAFETY: EraseMediaStoreError is repr(transparent) so T and + // EraseMediaStoreError have the same layout and ABI + unsafe { Arc::from_raw(ptr_erased) } + } +} diff --git a/crates/matrix-sdk-base/src/store/mod.rs b/crates/matrix-sdk-base/src/store/mod.rs index 1e5da059da5..93cd96fff77 100644 --- a/crates/matrix-sdk-base/src/store/mod.rs +++ b/crates/matrix-sdk-base/src/store/mod.rs @@ -72,6 +72,7 @@ use crate::{ MinimalRoomMemberEvent, Room, RoomCreateWithCreatorEventContent, RoomStateFilter, SessionMeta, deserialized_responses::DisplayName, event_cache::store as event_cache_store, + media::store as media_store, room::{RoomInfo, RoomInfoNotableUpdate, RoomState}, }; @@ -775,6 +776,7 @@ pub struct StoreConfig { pub(crate) crypto_store: Arc, pub(crate) state_store: Arc, pub(crate) event_cache_store: event_cache_store::EventCacheStoreLock, + pub(crate) media_store: media_store::MediaStoreLock, cross_process_store_locks_holder_name: String, } @@ -800,6 +802,10 @@ impl StoreConfig { event_cache_store::MemoryStore::new(), cross_process_store_locks_holder_name.clone(), ), + media_store: media_store::MediaStoreLock::new( + media_store::MemoryMediaStore::new(), + cross_process_store_locks_holder_name.clone(), + ), cross_process_store_locks_holder_name, } } @@ -830,6 +836,18 @@ impl StoreConfig { ); self } + + /// Set a custom implementation of an `MediaStore`. + pub fn media_store(mut self, media_store: S) -> Self + where + S: media_store::IntoMediaStore, + { + self.media_store = media_store::MediaStoreLock::new( + media_store, + self.cross_process_store_locks_holder_name.clone(), + ); + self + } } #[cfg(test)] diff --git a/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs b/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs index a9f4ab1630d..400112ce7e1 100644 --- a/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs +++ b/crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs @@ -18,10 +18,7 @@ use std::{rc::Rc, time::Duration}; use indexed_db_futures::IdbDatabase; use matrix_sdk_base::{ - event_cache::{ - store::EventCacheStore, - Event, Gap, - }, + event_cache::{store::EventCacheStore, Event, Gap}, linked_chunk::{ ChunkIdentifier, ChunkIdentifierGenerator, ChunkMetadata, LinkedChunkId, Position, RawChunk, Update, diff --git a/crates/matrix-sdk/src/client/builder/mod.rs b/crates/matrix-sdk/src/client/builder/mod.rs index abeb36c150b..409b5f5c54d 100644 --- a/crates/matrix-sdk/src/client/builder/mod.rs +++ b/crates/matrix-sdk/src/client/builder/mod.rs @@ -668,11 +668,20 @@ async fn build_store_config( .event_cache_store({ let mut config = config.clone(); - if let Some(cache_path) = cache_path { + if let Some(ref cache_path) = cache_path { config = config.path(cache_path); } matrix_sdk_sqlite::SqliteEventCacheStore::open_with_config(config).await? + }) + .media_store({ + let mut config = config.clone(); + + if let Some(ref cache_path) = cache_path { + config = config.path(cache_path); + } + + matrix_sdk_sqlite::SqliteMediaStore::open_with_config(config).await? }); #[cfg(feature = "e2e-encryption")] diff --git a/crates/matrix-sdk/src/client/mod.rs b/crates/matrix-sdk/src/client/mod.rs index e7c82f821a6..5e333b21687 100644 --- a/crates/matrix-sdk/src/client/mod.rs +++ b/crates/matrix-sdk/src/client/mod.rs @@ -32,6 +32,7 @@ use futures_util::StreamExt; use matrix_sdk_base::crypto::{store::LockableCryptoStore, DecryptionSettings}; use matrix_sdk_base::{ event_cache::store::EventCacheStoreLock, + media::store::MediaStoreLock, store::{DynStateStore, RoomLoadSettings, ServerInfo, WellKnownResponse}, sync::{Notification, RoomUpdates}, BaseClient, RoomInfoNotableUpdate, RoomState, RoomStateFilter, SendOutsideWasm, SessionMeta, @@ -741,6 +742,11 @@ impl Client { self.base_client().event_cache_store() } + /// Get a reference to the media store. + pub fn media_store(&self) -> &MediaStoreLock { + self.base_client().media_store() + } + /// Access the native Matrix authentication API with this client. pub fn matrix_auth(&self) -> MatrixAuth { MatrixAuth::new(self.clone()) diff --git a/crates/matrix-sdk/src/error.rs b/crates/matrix-sdk/src/error.rs index 34a05236bfc..3a9f9185023 100644 --- a/crates/matrix-sdk/src/error.rs +++ b/crates/matrix-sdk/src/error.rs @@ -25,8 +25,8 @@ use matrix_sdk_base::crypto::{ CryptoStoreError, DecryptorError, KeyExportError, MegolmError, OlmError, }; use matrix_sdk_base::{ - event_cache::store::EventCacheStoreError, Error as SdkBaseError, QueueWedgeError, RoomState, - StoreError, + event_cache::store::EventCacheStoreError, media::store::MediaStoreError, Error as SdkBaseError, + QueueWedgeError, RoomState, StoreError, }; use reqwest::Error as ReqwestError; use ruma::{ @@ -340,6 +340,10 @@ pub enum Error { #[error(transparent)] EventCacheStore(Box), + /// An error occurred in the media store. + #[error(transparent)] + MediaStore(Box), + /// An error encountered when trying to parse an identifier. #[error(transparent)] Identifier(#[from] IdParseError), @@ -507,6 +511,12 @@ impl From for Error { } } +impl From for Error { + fn from(error: MediaStoreError) -> Self { + Error::MediaStore(Box::new(error)) + } +} + #[cfg(feature = "qrcode")] impl From for Error { fn from(error: ScanError) -> Self { diff --git a/crates/matrix-sdk/src/lib.rs b/crates/matrix-sdk/src/lib.rs index 65dc912defd..aa01fdbc534 100644 --- a/crates/matrix-sdk/src/lib.rs +++ b/crates/matrix-sdk/src/lib.rs @@ -82,7 +82,8 @@ pub use http_client::TransmissionProgress; pub use matrix_sdk_sqlite::SqliteCryptoStore; #[cfg(feature = "sqlite")] pub use matrix_sdk_sqlite::{ - SqliteEventCacheStore, SqliteStateStore, SqliteStoreConfig, STATE_STORE_DATABASE_NAME, + SqliteEventCacheStore, SqliteMediaStore, SqliteStateStore, SqliteStoreConfig, + STATE_STORE_DATABASE_NAME, }; pub use media::Media; pub use pusher::Pusher; diff --git a/crates/matrix-sdk/src/media.rs b/crates/matrix-sdk/src/media.rs index 6bb41831bb6..b4016e3ed49 100644 --- a/crates/matrix-sdk/src/media.rs +++ b/crates/matrix-sdk/src/media.rs @@ -23,8 +23,8 @@ use std::{fmt, fs::File, path::Path}; use eyeball::SharedObservable; use futures_util::future::try_join; -use matrix_sdk_base::event_cache::store::media::IgnoreMediaRetentionPolicy; -pub use matrix_sdk_base::{event_cache::store::media::MediaRetentionPolicy, media::*}; +use matrix_sdk_base::media::store::IgnoreMediaRetentionPolicy; +pub use matrix_sdk_base::media::{store::MediaRetentionPolicy, *}; use mime::Mime; use ruma::{ api::{ @@ -428,7 +428,7 @@ impl Media { // Read from the cache. if use_cache { if let Some(content) = - self.client.event_cache_store().lock().await?.get_media_content(request).await? + self.client.media_store().lock().await?.get_media_content(request).await? { return Ok(content); } @@ -520,7 +520,7 @@ impl Media { if use_cache { self.client - .event_cache_store() + .media_store() .lock() .await? .add_media_content(request, content.clone(), IgnoreMediaRetentionPolicy::No) @@ -538,7 +538,7 @@ impl Media { async fn get_local_media_content(&self, uri: &MxcUri) -> Result> { // Read from the cache. self.client - .event_cache_store() + .media_store() .lock() .await? .get_media_content_for_uri(uri) @@ -552,7 +552,7 @@ impl Media { /// /// * `request` - The `MediaRequest` of the content. pub async fn remove_media_content(&self, request: &MediaRequestParameters) -> Result<()> { - Ok(self.client.event_cache_store().lock().await?.remove_media_content(request).await?) + Ok(self.client.media_store().lock().await?.remove_media_content(request).await?) } /// Delete all the media content corresponding to the given @@ -562,7 +562,7 @@ impl Media { /// /// * `uri` - The `MxcUri` of the files. pub async fn remove_media_content_for_uri(&self, uri: &MxcUri) -> Result<()> { - Ok(self.client.event_cache_store().lock().await?.remove_media_content_for_uri(uri).await?) + Ok(self.client.media_store().lock().await?.remove_media_content_for_uri(uri).await?) } /// Get the file of the given media event content. @@ -697,20 +697,20 @@ impl Media { /// /// * `policy` - The `MediaRetentionPolicy` to use. pub async fn set_media_retention_policy(&self, policy: MediaRetentionPolicy) -> Result<()> { - self.client.event_cache_store().lock().await?.set_media_retention_policy(policy).await?; + 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 { - Ok(self.client.event_cache_store().lock().await?.media_retention_policy()) + 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_up_media_cache(&self) -> Result<()> { - self.client.event_cache_store().lock().await?.clean_up_media_cache().await?; + self.client.media_store().lock().await?.clean_up_media_cache().await?; Ok(()) } diff --git a/crates/matrix-sdk/src/room/mod.rs b/crates/matrix-sdk/src/room/mod.rs index b04edce7341..9e82d33c008 100644 --- a/crates/matrix-sdk/src/room/mod.rs +++ b/crates/matrix-sdk/src/room/mod.rs @@ -43,8 +43,7 @@ use matrix_sdk_base::{ deserialized_responses::{ RawAnySyncOrStrippedState, RawSyncOrStrippedState, SyncOrStrippedState, }, - event_cache::store::media::IgnoreMediaRetentionPolicy, - media::MediaThumbnailSettings, + media::{store::IgnoreMediaRetentionPolicy, MediaThumbnailSettings}, store::{StateStoreExt, ThreadSubscriptionStatus}, ComposerDraft, EncryptionState, RoomInfoNotableUpdateReasons, RoomMemberships, SendOutsideWasm, StateChanges, StateStoreDataKey, StateStoreDataValue, @@ -2582,7 +2581,7 @@ impl Room { .await?; if store_in_cache { - let cache_store_lock_guard = self.client.event_cache_store().lock().await?; + let media_store_lock_guard = self.client.media_store().lock().await?; // A failure to cache shouldn't prevent the whole upload from finishing // properly, so only log errors during caching. @@ -2591,7 +2590,7 @@ impl Room { let request = MediaRequestParameters { source: media_source.clone(), format: MediaFormat::File }; - if let Err(err) = cache_store_lock_guard + if let Err(err) = media_store_lock_guard .add_media_content(&request, data, IgnoreMediaRetentionPolicy::No) .await { @@ -2608,7 +2607,7 @@ impl Room { format: MediaFormat::Thumbnail(MediaThumbnailSettings::new(width, height)), }; - if let Err(err) = cache_store_lock_guard + if let Err(err) = media_store_lock_guard .add_media_content(&request, data, IgnoreMediaRetentionPolicy::No) .await { diff --git a/crates/matrix-sdk/src/send_queue/mod.rs b/crates/matrix-sdk/src/send_queue/mod.rs index d7146031041..03eef775eb8 100644 --- a/crates/matrix-sdk/src/send_queue/mod.rs +++ b/crates/matrix-sdk/src/send_queue/mod.rs @@ -142,7 +142,7 @@ use eyeball::SharedObservable; use matrix_sdk_base::store::FinishGalleryItemInfo; use matrix_sdk_base::{ event_cache::store::EventCacheStoreError, - media::MediaRequestParameters, + media::{store::MediaStoreError, MediaRequestParameters}, store::{ ChildTransactionId, DependentQueuedRequest, DependentQueuedRequestKind, DynStateStore, FinishUploadThumbnailInfo, QueueWedgeError, QueuedRequest, QueuedRequestKind, @@ -842,7 +842,7 @@ impl RoomSendQueue { let fut = async move { let data = room .client() - .event_cache_store() + .media_store() .lock() .await? .get_media_content(&cache_key) @@ -2348,6 +2348,10 @@ pub enum RoomSendQueueStorageError { #[error(transparent)] EventCacheStoreError(#[from] EventCacheStoreError), + /// Error caused by the event cache store. + #[error(transparent)] + MediaStoreError(#[from] MediaStoreError), + /// Error caused when attempting to get a handle on the event cache store. #[error(transparent)] LockError(#[from] LockStoreError), diff --git a/crates/matrix-sdk/src/send_queue/progress.rs b/crates/matrix-sdk/src/send_queue/progress.rs index 72fa1b900bc..11a0de2fab0 100644 --- a/crates/matrix-sdk/src/send_queue/progress.rs +++ b/crates/matrix-sdk/src/send_queue/progress.rs @@ -94,17 +94,17 @@ impl RoomSendQueue { }; // Get the size of the file being uploaded from the event cache. - let bytes = match room.client().event_cache_store().lock().await { + let bytes = match room.client().media_store().lock().await { Ok(cache) => match cache.get_media_content(cache_key).await { Ok(Some(content)) => content.len(), Ok(None) => 0, Err(err) => { - warn!("error when reading media content from cache store: {err}"); + warn!("error when reading media content from media store: {err}"); 0 } }, Err(err) => { - warn!("couldn't acquire cache store lock: {err}"); + warn!("couldn't acquire media store lock: {err}"); 0 } }; @@ -195,9 +195,9 @@ impl RoomSendQueue { return Ok(None); } - let cache_store_guard = client.event_cache_store().lock().await?; + let media_store_guard = client.media_store().lock().await?; - let maybe_content = cache_store_guard.get_media_content(&cache_key).await?; + let maybe_content = media_store_guard.get_media_content(&cache_key).await?; Ok(maybe_content.map(|c| c.len())) } diff --git a/crates/matrix-sdk/src/send_queue/upload.rs b/crates/matrix-sdk/src/send_queue/upload.rs index c12b5d496bb..30ec4ca0099 100644 --- a/crates/matrix-sdk/src/send_queue/upload.rs +++ b/crates/matrix-sdk/src/send_queue/upload.rs @@ -17,20 +17,19 @@ #[cfg(feature = "unstable-msc4274")] use std::{collections::HashMap, iter::zip}; +#[cfg(feature = "unstable-msc4274")] +use matrix_sdk_base::{ + media::UniqueKey, + store::{AccumulatedSentMediaInfo, FinishGalleryItemInfo}, +}; use matrix_sdk_base::{ - event_cache::store::media::IgnoreMediaRetentionPolicy, - media::{MediaFormat, MediaRequestParameters}, + media::{store::IgnoreMediaRetentionPolicy, MediaFormat, MediaRequestParameters}, store::{ ChildTransactionId, DependentQueuedRequestKind, FinishUploadThumbnailInfo, QueuedRequestKind, SentMediaInfo, SentRequestKey, SerializableEventContent, }, RoomState, }; -#[cfg(feature = "unstable-msc4274")] -use matrix_sdk_base::{ - media::UniqueKey, - store::{AccumulatedSentMediaInfo, FinishGalleryItemInfo}, -}; use mime::Mime; #[cfg(feature = "unstable-msc4274")] use ruma::events::room::message::{GalleryItemType, GalleryMessageEventContent}; @@ -420,14 +419,11 @@ impl RoomSendQueue { file_media_request: &MediaRequestParameters, ) -> Result { let client = room.client(); - let cache_store = client - .event_cache_store() - .lock() - .await - .map_err(RoomSendQueueStorageError::LockError)?; + let media_store = + client.media_store().lock().await.map_err(RoomSendQueueStorageError::LockError)?; // Cache the file itself in the cache store. - cache_store + media_store .add_media_content( file_media_request, data, @@ -435,7 +431,7 @@ impl RoomSendQueue { IgnoreMediaRetentionPolicy::Yes, ) .await - .map_err(RoomSendQueueStorageError::EventCacheStoreError)?; + .map_err(RoomSendQueueStorageError::MediaStoreError)?; // Process the thumbnail, if it's been provided. if let Some(thumbnail) = thumbnail { @@ -449,7 +445,7 @@ impl RoomSendQueue { // Cache thumbnail in the cache store. let thumbnail_media_request = Media::make_local_file_media_request(&txn); - cache_store + media_store .add_media_content( &thumbnail_media_request, data, @@ -457,7 +453,7 @@ impl RoomSendQueue { IgnoreMediaRetentionPolicy::Yes, ) .await - .map_err(RoomSendQueueStorageError::EventCacheStoreError)?; + .map_err(RoomSendQueueStorageError::MediaStoreError)?; Ok(MediaCacheResult { upload_thumbnail_txn: Some(txn.clone()), @@ -793,12 +789,12 @@ impl QueueStorage { // At this point, all the requests and dependent requests have been cleaned up. // Perform the final step: empty the cache from the local items. { - let event_cache = client.event_cache_store().lock().await?; - event_cache + let media_store = client.media_store().lock().await?; + media_store .remove_media_content_for_uri(&Media::make_local_uri(&handles.upload_file_txn)) .await?; if let Some(txn) = &handles.upload_thumbnail_txn { - event_cache.remove_media_content_for_uri(&Media::make_local_uri(txn)).await?; + media_store.remove_media_content_for_uri(&Media::make_local_uri(txn)).await?; } } @@ -938,22 +934,22 @@ async fn update_media_cache_keys_after_upload( let from_req = Media::make_local_file_media_request(file_upload_txn); trace!(from = ?from_req.source, to = ?sent_media.file, "renaming media file key in cache store"); - let cache_store = - client.event_cache_store().lock().await.map_err(RoomSendQueueStorageError::LockError)?; + let media_store = + client.media_store().lock().await.map_err(RoomSendQueueStorageError::LockError)?; // The media can now be removed during cleanups. - cache_store + media_store .set_ignore_media_retention_policy(&from_req, IgnoreMediaRetentionPolicy::No) .await - .map_err(RoomSendQueueStorageError::EventCacheStoreError)?; + .map_err(RoomSendQueueStorageError::MediaStoreError)?; - cache_store + media_store .replace_media_key( &from_req, &MediaRequestParameters { source: sent_media.file.clone(), format: MediaFormat::File }, ) .await - .map_err(RoomSendQueueStorageError::EventCacheStoreError)?; + .map_err(RoomSendQueueStorageError::MediaStoreError)?; // Rename the thumbnail too, if needs be. if let Some((info, new_source)) = thumbnail_info.as_ref().zip(sent_media.thumbnail.clone()) { @@ -968,18 +964,18 @@ async fn update_media_cache_keys_after_upload( trace!(from = ?from_req.source, to = ?new_source, "renaming thumbnail file key in cache store"); // The media can now be removed during cleanups. - cache_store + media_store .set_ignore_media_retention_policy(&from_req, IgnoreMediaRetentionPolicy::No) .await - .map_err(RoomSendQueueStorageError::EventCacheStoreError)?; + .map_err(RoomSendQueueStorageError::MediaStoreError)?; - cache_store + media_store .replace_media_key( &from_req, &MediaRequestParameters { source: new_source, format: MediaFormat::File }, ) .await - .map_err(RoomSendQueueStorageError::EventCacheStoreError)?; + .map_err(RoomSendQueueStorageError::MediaStoreError)?; } Ok(())