matrix_sdk_base/media/store/
media_service.rs

1// Copyright 2025 Kévin Commaille
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
15use std::sync::Arc;
16
17use matrix_sdk_common::{
18    SendOutsideWasm, SyncOutsideWasm,
19    executor::{JoinHandle, spawn},
20    locks::Mutex,
21};
22use ruma::time::SystemTime;
23use tokio::sync::Mutex as AsyncMutex;
24use tracing::error;
25
26use super::{MediaRetentionPolicy, MediaStoreInner};
27use crate::media::MediaRequestParameters;
28
29/// API for implementors of [`MediaStore`] to manage their media through their
30/// implementation of [`MediaStoreInner`].
31///
32/// [`MediaStore`]: crate::media::store::MediaStore
33#[derive(Debug)]
34pub struct MediaService<Time: TimeProvider = DefaultTimeProvider> {
35    inner: Arc<MediaServiceInner<Time>>,
36}
37
38#[derive(Debug)]
39struct MediaServiceInner<Time: TimeProvider = DefaultTimeProvider> {
40    /// The time provider.
41    time_provider: Time,
42
43    /// The current [`MediaRetentionPolicy`].
44    policy: Mutex<MediaRetentionPolicy>,
45
46    /// A mutex to ensure a single cleanup is running at a time.
47    cleanup_guard: AsyncMutex<()>,
48
49    /// The time of the last media cache cleanup.
50    last_media_cleanup_time: Mutex<Option<SystemTime>>,
51
52    /// The [`JoinHandle`] for an automatic media cleanup task.
53    ///
54    /// Used to ensure that only one automatic cleanup is running at a time, and
55    /// to stop the cleanup when the [`MediaServiceInner`] is dropped.
56    automatic_media_cleanup_join_handle: Mutex<Option<JoinHandle<()>>>,
57}
58
59impl MediaService {
60    /// Construct a new default `MediaService`.
61    ///
62    /// [`MediaService::restore()`] should be called after constructing the
63    /// `MediaService` to restore its previous state.
64    pub fn new() -> Self {
65        Self::default()
66    }
67}
68
69impl Default for MediaService {
70    fn default() -> Self {
71        Self::with_time_provider(DefaultTimeProvider)
72    }
73}
74
75impl<Time> MediaService<Time>
76where
77    Time: TimeProvider + 'static,
78{
79    /// Construct a new `MediaService` with the given `TimeProvider` and an
80    /// empty `MediaRetentionPolicy`.
81    fn with_time_provider(time_provider: Time) -> Self {
82        let inner = MediaServiceInner {
83            time_provider,
84            policy: Mutex::new(MediaRetentionPolicy::empty()),
85            cleanup_guard: AsyncMutex::new(()),
86            last_media_cleanup_time: Mutex::new(None),
87            automatic_media_cleanup_join_handle: Mutex::new(None),
88        };
89
90        Self { inner: Arc::new(inner) }
91    }
92
93    /// Restore the previous state of the [`MediaRetentionPolicy`] from data
94    /// that was persisted in the store.
95    ///
96    /// This should be called immediately after constructing the `MediaService`.
97    ///
98    /// # Arguments
99    ///
100    /// - `policy` - The `MediaRetentionPolicy` that was persisted in the store.
101    pub fn restore(
102        &self,
103        policy: Option<MediaRetentionPolicy>,
104        last_media_cleanup_time: Option<SystemTime>,
105    ) {
106        if let Some(policy) = policy {
107            *self.inner.policy.lock() = policy;
108        }
109
110        if let Some(time) = last_media_cleanup_time {
111            *self.inner.last_media_cleanup_time.lock() = Some(time);
112        }
113    }
114
115    /// Get the current time from the inner [`TimeProvider`].
116    fn now(&self) -> SystemTime {
117        self.inner.time_provider.now()
118    }
119
120    /// Set the `MediaRetentionPolicy` of this service.
121    ///
122    /// # Arguments
123    ///
124    /// * `store` - The `MediaStoreInner`.
125    ///
126    /// * `policy` - The `MediaRetentionPolicy` to use.
127    pub async fn set_media_retention_policy<Store: MediaStoreInner + 'static>(
128        &self,
129        store: &Store,
130        policy: MediaRetentionPolicy,
131    ) -> Result<(), Store::Error> {
132        store.set_media_retention_policy_inner(policy).await?;
133
134        *self.inner.policy.lock() = policy;
135
136        self.maybe_spawn_automatic_media_cache_cleanup(store, self.now());
137
138        Ok(())
139    }
140
141    /// Get the `MediaRetentionPolicy` of this service.
142    pub fn media_retention_policy(&self) -> MediaRetentionPolicy {
143        *self.inner.policy.lock()
144    }
145
146    /// Add a media file's content in the media store.
147    ///
148    /// # Arguments
149    ///
150    /// - `store` - The `MediaStoreInner`.
151    /// - `request` - The `MediaRequestParameters` of the file.
152    /// - `content` - The content of the file.
153    /// - `ignore_policy` - Whether the current `MediaRetentionPolicy` should be
154    ///   ignored.
155    pub async fn add_media_content<Store: MediaStoreInner + 'static>(
156        &self,
157        store: &Store,
158        request: &MediaRequestParameters,
159        content: Vec<u8>,
160        ignore_policy: IgnoreMediaRetentionPolicy,
161    ) -> Result<(), Store::Error> {
162        let policy = self.media_retention_policy();
163
164        if ignore_policy == IgnoreMediaRetentionPolicy::No
165            && policy.exceeds_max_file_size(content.len() as u64)
166        {
167            // We do not cache the content.
168            return Ok(());
169        }
170
171        let current_time = self.now();
172        store
173            .add_media_content_inner(request, content, current_time, policy, ignore_policy)
174            .await?;
175
176        self.maybe_spawn_automatic_media_cache_cleanup(store, current_time);
177
178        Ok(())
179    }
180
181    /// Set whether the current [`MediaRetentionPolicy`] should be ignored for
182    /// the media.
183    ///
184    /// The change will be taken into account in the next cleanup.
185    ///
186    /// # Arguments
187    ///
188    /// - `store` - The `MediaStoreInner`.
189    /// - `request` - The `MediaRequestParameters` of the file.
190    /// - `ignore_policy` - Whether the current `MediaRetentionPolicy` should be
191    ///   ignored.
192    pub async fn set_ignore_media_retention_policy<Store: MediaStoreInner>(
193        &self,
194        store: &Store,
195        request: &MediaRequestParameters,
196        ignore_policy: IgnoreMediaRetentionPolicy,
197    ) -> Result<(), Store::Error> {
198        store.set_ignore_media_retention_policy_inner(request, ignore_policy).await
199    }
200
201    /// Get a media file's content out of the media store.
202    ///
203    /// # Arguments
204    ///
205    /// * `store` - The `MediaStoreInner`.
206    ///
207    /// * `request` - The `MediaRequestParameters` of the file.
208    pub async fn get_media_content<Store: MediaStoreInner + 'static>(
209        &self,
210        store: &Store,
211        request: &MediaRequestParameters,
212    ) -> Result<Option<Vec<u8>>, Store::Error> {
213        let current_time = self.now();
214        let content = store.get_media_content_inner(request, current_time).await?;
215
216        self.maybe_spawn_automatic_media_cache_cleanup(store, current_time);
217
218        Ok(content)
219    }
220
221    /// Clean up the media cache with the current `MediaRetentionPolicy`.
222    ///
223    /// If there is already an ongoing cleanup, this is a noop.
224    ///
225    /// # Arguments
226    ///
227    /// * `store` - The `MediaStoreInner`.
228    pub async fn clean<Store: MediaStoreInner>(&self, store: &Store) -> Result<(), Store::Error> {
229        self.clean_inner(store, self.now()).await
230    }
231
232    async fn clean_inner<Store: MediaStoreInner>(
233        &self,
234        store: &Store,
235        current_time: SystemTime,
236    ) -> Result<(), Store::Error> {
237        let Ok(_guard) = self.inner.cleanup_guard.try_lock() else {
238            // There is another ongoing cleanup.
239            return Ok(());
240        };
241
242        let policy = self.media_retention_policy();
243
244        if !policy.has_limitations() {
245            // No need to call the backend.
246            return Ok(());
247        }
248
249        store.clean_inner(policy, current_time).await?;
250
251        *self.inner.last_media_cleanup_time.lock() = Some(current_time);
252
253        Ok(())
254    }
255
256    /// Spawn an automatic media cache cleanup, according to the media retention
257    /// policy.
258    ///
259    /// A cleanup will be spawned if:
260    ///
261    /// - The media retention policy's `cleanup_frequency` is set and enough
262    ///   time has passed since the last cleanup.
263    /// - No other cleanup is running,
264    fn maybe_spawn_automatic_media_cache_cleanup<Store: MediaStoreInner + 'static>(
265        &self,
266        store: &Store,
267        current_time: SystemTime,
268    ) {
269        let mut join_handle = self.inner.automatic_media_cleanup_join_handle.lock();
270
271        if join_handle.as_ref().is_some_and(|join_handle| !join_handle.is_finished()) {
272            // There is an ongoing automatic media cache cleanup.
273            return;
274        }
275
276        let policy = self.media_retention_policy();
277        if policy.cleanup_frequency.is_none() || !policy.has_limitations() {
278            // Automatic cleanups are disabled or have no effect.
279            return;
280        }
281
282        let last_media_cleanup_time = *self.inner.last_media_cleanup_time.lock();
283        if last_media_cleanup_time.is_some_and(|last_cleanup_time| {
284            !policy.should_clean_up(current_time, last_cleanup_time)
285        }) {
286            // It is not time to clean up.
287            return;
288        }
289
290        let this = self.clone();
291        let store = store.clone();
292
293        let handle = spawn(async move {
294            if let Err(error) = this.clean_inner(&store, current_time).await {
295                error!("Failed to run automatic media cache cleanup: {error}");
296            }
297        });
298
299        *join_handle = Some(handle);
300    }
301}
302
303impl<Time> Clone for MediaService<Time>
304where
305    Time: TimeProvider,
306{
307    fn clone(&self) -> Self {
308        Self { inner: self.inner.clone() }
309    }
310}
311
312impl<Time> Drop for MediaServiceInner<Time>
313where
314    Time: TimeProvider,
315{
316    fn drop(&mut self) {
317        if let Some(join_handle) = self.automatic_media_cleanup_join_handle.lock().take() {
318            join_handle.abort();
319        }
320    }
321}
322
323/// Whether the [`MediaRetentionPolicy`] should be ignored for the current
324/// content.
325///
326/// Some media cache actions are noops when the media content that is processed
327/// is filtered out by the policy. This can break some features of the SDK, like
328/// the send queue, that expects to be able to persist all media files in the
329/// store to restore them when the client is restored.
330///
331/// This can be converted to a boolean with
332/// [`IgnoreMediaRetentionPolicy::is_yes()`].
333#[derive(Debug, Clone, Copy, PartialEq, Eq)]
334pub enum IgnoreMediaRetentionPolicy {
335    /// The media retention policy will be ignored and the current action will
336    /// not be a noop.
337    ///
338    /// Any media content in this state must NOT be used when applying a
339    /// `MediaRetentionPolicy`. This applies to ANY criteria, like the maximum
340    /// file size, the maximum cache size or the last access expiry.
341    ///
342    /// This state is supposed to be transient, and to only be used internally
343    /// by the SDK.
344    Yes,
345
346    /// The media retention policy will be respected and the current action
347    /// might be a noop.
348    No,
349}
350
351impl IgnoreMediaRetentionPolicy {
352    /// Whether this is an [`IgnoreMediaRetentionPolicy::Yes`] variant.
353    pub fn is_yes(self) -> bool {
354        matches!(self, Self::Yes)
355    }
356}
357
358/// An abstract trait to provide the current `SystemTime` for the
359/// [`MediaService`].
360pub trait TimeProvider: SendOutsideWasm + SyncOutsideWasm {
361    /// The current time.
362    fn now(&self) -> SystemTime;
363}
364
365/// The default time provider, that calls `ruma::time::SystemTime::now()`.
366#[derive(Debug)]
367pub struct DefaultTimeProvider;
368
369impl TimeProvider for DefaultTimeProvider {
370    fn now(&self) -> SystemTime {
371        SystemTime::now()
372    }
373}
374
375#[cfg(test)]
376mod tests {
377    use std::{
378        fmt,
379        sync::{Arc, MutexGuard},
380    };
381
382    use async_trait::async_trait;
383    use matrix_sdk_common::locks::Mutex;
384    use matrix_sdk_test::async_test;
385    use ruma::{
386        OwnedMxcUri,
387        events::room::MediaSource,
388        mxc_uri,
389        time::{Duration, SystemTime},
390    };
391
392    use super::{
393        IgnoreMediaRetentionPolicy, MediaRetentionPolicy, MediaService, MediaStoreInner,
394        TimeProvider,
395    };
396    use crate::media::{MediaFormat, MediaRequestParameters, UniqueKey, store::MediaStoreError};
397
398    #[derive(Debug, Default, Clone)]
399    struct MockMediaStoreInner {
400        inner: Arc<Mutex<MockMediaStoreInnerInner>>,
401    }
402
403    impl MockMediaStoreInner {
404        /// Whether the store was accessed.
405        fn accessed(&self) -> bool {
406            self.inner.lock().accessed
407        }
408
409        /// Reset the `accessed` boolean.
410        fn reset_accessed(&self) {
411            self.inner.lock().accessed = false;
412        }
413
414        /// Access the inner store.
415        ///
416        /// Should be called for every access to the inner store as it also sets
417        /// the `accessed` boolean.
418        fn inner(&self) -> MutexGuard<'_, MockMediaStoreInnerInner> {
419            let mut inner = self.inner.lock();
420            inner.accessed = true;
421            inner
422        }
423    }
424
425    #[derive(Debug, Default)]
426    struct MockMediaStoreInnerInner {
427        /// Whether this store was accessed.
428        ///
429        /// Must be set to `true` for any operation that unlocks the store.
430        accessed: bool,
431
432        /// The persisted media retention policy.
433        media_retention_policy: Option<MediaRetentionPolicy>,
434
435        /// The list of media content.
436        media_list: Vec<MediaContent>,
437
438        /// The time of the last cleanup.
439        cleanup_time: Option<SystemTime>,
440    }
441
442    #[derive(Debug, Clone)]
443    struct MediaContent {
444        /// The unique key for the media content.
445        key: String,
446
447        /// The original URI of the media content.
448        uri: OwnedMxcUri,
449
450        /// The media content.
451        content: Vec<u8>,
452
453        /// Whether the `MediaRetentionPolicy` should be ignored for this media
454        /// content;
455        ignore_policy: bool,
456
457        /// The time of the last access of the media content.
458        last_access: SystemTime,
459    }
460
461    #[derive(Debug)]
462    struct MockMediaStoreInnerError;
463
464    impl fmt::Display for MockMediaStoreInnerError {
465        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
466            write!(f, "MockMediaStoreInnerError")
467        }
468    }
469
470    impl std::error::Error for MockMediaStoreInnerError {}
471
472    impl From<MockMediaStoreInnerError> for MediaStoreError {
473        fn from(value: MockMediaStoreInnerError) -> Self {
474            Self::backend(value)
475        }
476    }
477
478    #[cfg_attr(target_family = "wasm", async_trait(?Send))]
479    #[cfg_attr(not(target_family = "wasm"), async_trait)]
480    impl MediaStoreInner for MockMediaStoreInner {
481        type Error = MockMediaStoreInnerError;
482
483        async fn media_retention_policy_inner(
484            &self,
485        ) -> Result<Option<MediaRetentionPolicy>, Self::Error> {
486            Ok(self.inner().media_retention_policy)
487        }
488
489        async fn set_media_retention_policy_inner(
490            &self,
491            policy: MediaRetentionPolicy,
492        ) -> Result<(), Self::Error> {
493            self.inner().media_retention_policy = Some(policy);
494            Ok(())
495        }
496
497        async fn add_media_content_inner(
498            &self,
499            request: &MediaRequestParameters,
500            content: Vec<u8>,
501            current_time: SystemTime,
502            policy: MediaRetentionPolicy,
503            ignore_policy: IgnoreMediaRetentionPolicy,
504        ) -> Result<(), Self::Error> {
505            let ignore_policy = ignore_policy.is_yes();
506
507            if !ignore_policy && policy.exceeds_max_file_size(content.len() as u64) {
508                return Ok(());
509            }
510
511            let mut inner = self.inner();
512            let key = request.unique_key();
513
514            if let Some(pos) = inner.media_list.iter().position(|content| content.key == key) {
515                let media_content = &mut inner.media_list[pos];
516                media_content.content = content;
517                media_content.last_access = current_time;
518                media_content.ignore_policy = ignore_policy;
519            } else {
520                inner.media_list.push(MediaContent {
521                    key,
522                    uri: request.uri().to_owned(),
523                    content,
524                    ignore_policy,
525                    last_access: current_time,
526                });
527            }
528
529            Ok(())
530        }
531
532        async fn set_ignore_media_retention_policy_inner(
533            &self,
534            request: &MediaRequestParameters,
535            ignore_policy: IgnoreMediaRetentionPolicy,
536        ) -> Result<(), Self::Error> {
537            let key = request.unique_key();
538            let mut inner = self.inner();
539
540            if let Some(pos) = inner.media_list.iter().position(|content| content.key == key) {
541                inner.media_list[pos].ignore_policy = ignore_policy.is_yes();
542            }
543
544            Ok(())
545        }
546
547        async fn get_media_content_inner(
548            &self,
549            request: &MediaRequestParameters,
550            current_time: SystemTime,
551        ) -> Result<Option<Vec<u8>>, Self::Error> {
552            let key = request.unique_key();
553            let mut inner = self.inner();
554
555            let Some(media_content) =
556                inner.media_list.iter_mut().find(|content| content.key == key)
557            else {
558                return Ok(None);
559            };
560
561            media_content.last_access = current_time;
562
563            Ok(Some(media_content.content.clone()))
564        }
565
566        async fn clean_inner(
567            &self,
568            _policy: MediaRetentionPolicy,
569            current_time: SystemTime,
570        ) -> Result<(), Self::Error> {
571            // This is mostly a noop. We don't care about this test
572            // implementation, only whether this method was called with the
573            // right time.
574            self.inner().cleanup_time = Some(current_time);
575
576            Ok(())
577        }
578
579        async fn last_media_cleanup_time_inner(&self) -> Result<Option<SystemTime>, Self::Error> {
580            Ok(self.inner().cleanup_time)
581        }
582    }
583
584    #[derive(Debug)]
585    struct MockTimeProvider {
586        now: Mutex<SystemTime>,
587    }
588
589    impl MockTimeProvider {
590        /// Construct a `MockTimeProvider` with the given current time.
591        fn new(now: SystemTime) -> Self {
592            Self { now: Mutex::new(now) }
593        }
594
595        /// Set the current time.
596        fn set_now(&self, now: SystemTime) {
597            *self.now.lock() = now;
598        }
599    }
600
601    impl TimeProvider for MockTimeProvider {
602        fn now(&self) -> SystemTime {
603            *self.now.lock()
604        }
605    }
606
607    #[async_test]
608    async fn test_media_service_empty_policy() {
609        let content = b"some text content";
610        let uri = mxc_uri!("mxc://server.local/AbcDe1234");
611        let request = MediaRequestParameters {
612            source: MediaSource::Plain(uri.to_owned()),
613            format: MediaFormat::File,
614        };
615
616        let now = SystemTime::UNIX_EPOCH;
617
618        let store = MockMediaStoreInner::default();
619        let service = MediaService::with_time_provider(MockTimeProvider::new(now));
620
621        // By default an empty policy is used.
622        assert!(!service.media_retention_policy().has_limitations());
623        service.restore(None, None);
624        assert!(!service.media_retention_policy().has_limitations());
625        assert!(!store.accessed());
626
627        // Add media.
628        service
629            .add_media_content(&store, &request, content.to_vec(), IgnoreMediaRetentionPolicy::No)
630            .await
631            .unwrap();
632        assert!(store.accessed());
633
634        let media_content = store.inner().media_list[0].clone();
635        assert_eq!(media_content.uri, uri);
636        assert_eq!(media_content.content, content);
637        assert!(!media_content.ignore_policy);
638        assert_eq!(media_content.last_access, now);
639
640        let now = now + Duration::from_secs(60);
641        service.inner.time_provider.set_now(now);
642        store.reset_accessed();
643
644        // Get media from request.
645        let loaded_content = service.get_media_content(&store, &request).await.unwrap();
646        assert!(store.accessed());
647        assert_eq!(loaded_content.as_deref(), Some(content.as_slice()));
648
649        // The last access time was updated.
650        let media = store.inner().media_list[0].clone();
651        assert_eq!(media.last_access, now);
652
653        let now = now + Duration::from_secs(60);
654        service.inner.time_provider.set_now(now);
655        store.reset_accessed();
656
657        // Update ignore_policy.
658        service
659            .set_ignore_media_retention_policy(&store, &request, IgnoreMediaRetentionPolicy::Yes)
660            .await
661            .unwrap();
662        assert!(store.accessed());
663
664        let media_content = store.inner().media_list[0].clone();
665        assert!(media_content.ignore_policy);
666
667        // Try a cleanup. With the empty policy the store should not be
668        // accessed.
669        assert_eq!(store.last_media_cleanup_time_inner().await.unwrap(), None);
670        store.reset_accessed();
671
672        service.clean(&store).await.unwrap();
673        assert!(!store.accessed());
674        assert_eq!(store.last_media_cleanup_time_inner().await.unwrap(), None);
675    }
676
677    #[async_test]
678    async fn test_media_service_non_empty_policy() {
679        // Content of less than 32 bytes.
680        let small_content = b"some text content";
681        let small_uri = mxc_uri!("mxc://server.local/small");
682        let small_request = MediaRequestParameters {
683            source: MediaSource::Plain(small_uri.to_owned()),
684            format: MediaFormat::File,
685        };
686
687        // Content of more than 32 bytes.
688        let big_content = b"some much much larger text content";
689        let big_uri = mxc_uri!("mxc://server.local/big");
690        let big_request = MediaRequestParameters {
691            source: MediaSource::Plain(big_uri.to_owned()),
692            format: MediaFormat::File,
693        };
694
695        // Limit the file size to 32 bytes in the retention policy.
696        let policy = MediaRetentionPolicy { max_file_size: Some(32), ..Default::default() };
697
698        let now = SystemTime::UNIX_EPOCH;
699
700        let store = MockMediaStoreInner::default();
701        let service = MediaService::with_time_provider(MockTimeProvider::new(now));
702
703        // Check that restoring the policy works.
704        service.restore(Some(MediaRetentionPolicy::default()), None);
705        assert_eq!(service.media_retention_policy(), MediaRetentionPolicy::default());
706        assert!(!store.accessed());
707
708        // Set the media retention policy.
709        service.set_media_retention_policy(&store, policy).await.unwrap();
710        assert!(store.accessed());
711        assert_eq!(service.media_retention_policy(), policy);
712        assert_eq!(store.inner().media_retention_policy, Some(policy));
713
714        store.reset_accessed();
715
716        // Add small media, it should work because its size is lower than the
717        // max file size.
718        service
719            .add_media_content(
720                &store,
721                &small_request,
722                small_content.to_vec(),
723                IgnoreMediaRetentionPolicy::No,
724            )
725            .await
726            .unwrap();
727        assert!(store.accessed());
728
729        let media_content = store.inner().media_list[0].clone();
730        assert_eq!(media_content.uri, small_uri);
731        assert_eq!(media_content.content, small_content);
732        assert!(!media_content.ignore_policy);
733        assert_eq!(media_content.last_access, now);
734
735        let now = now + Duration::from_secs(60);
736        service.inner.time_provider.set_now(now);
737        store.reset_accessed();
738
739        // Get media from request.
740        let loaded_content = service.get_media_content(&store, &small_request).await.unwrap();
741        assert!(store.accessed());
742        assert_eq!(loaded_content.as_deref(), Some(small_content.as_slice()));
743
744        // The last access time was updated.
745        let media = store.inner().media_list[0].clone();
746        assert_eq!(media.last_access, now);
747
748        let now = now + Duration::from_secs(60);
749        service.inner.time_provider.set_now(now);
750        store.reset_accessed();
751
752        let now = now + Duration::from_secs(60);
753        service.inner.time_provider.set_now(now);
754        store.reset_accessed();
755
756        // Add big media, it will not work because it is bigger than the max
757        // file size.
758        service
759            .add_media_content(
760                &store,
761                &big_request,
762                big_content.to_vec(),
763                IgnoreMediaRetentionPolicy::No,
764            )
765            .await
766            .unwrap();
767        assert!(!store.accessed());
768        assert_eq!(store.inner().media_list.len(), 1);
769
770        store.reset_accessed();
771
772        let loaded_content = service.get_media_content(&store, &big_request).await.unwrap();
773        assert!(store.accessed());
774        assert_eq!(loaded_content, None);
775
776        store.reset_accessed();
777
778        // Add big media, but this time ignore the policy.
779        service
780            .add_media_content(
781                &store,
782                &big_request,
783                big_content.to_vec(),
784                IgnoreMediaRetentionPolicy::Yes,
785            )
786            .await
787            .unwrap();
788        assert!(store.accessed());
789        assert_eq!(store.inner().media_list.len(), 2);
790
791        store.reset_accessed();
792
793        // Get media from request.
794        let loaded_content = service.get_media_content(&store, &big_request).await.unwrap();
795        assert!(store.accessed());
796        assert_eq!(loaded_content.as_deref(), Some(big_content.as_slice()));
797
798        // The last access time was updated.
799        let media = store.inner().media_list[1].clone();
800        assert_eq!(media.last_access, now);
801
802        // Try a cleanup, the store should be accessed.
803        assert_eq!(store.last_media_cleanup_time_inner().await.unwrap(), None);
804
805        let now = now + Duration::from_secs(60);
806        service.inner.time_provider.set_now(now);
807        store.reset_accessed();
808
809        service.clean(&store).await.unwrap();
810        assert!(store.accessed());
811        assert_eq!(store.last_media_cleanup_time_inner().await.unwrap(), Some(now));
812    }
813
814    #[async_test]
815    async fn test_media_service_automatic_cleanup() {
816        // 64 bytes content.
817        let content = vec![0; 64];
818
819        let uri_1 = mxc_uri!("mxc://localhost/media-1");
820        let request_1 = MediaRequestParameters {
821            source: MediaSource::Plain(uri_1.to_owned()),
822            format: MediaFormat::File,
823        };
824        let uri_2 = mxc_uri!("mxc://localhost/media-2");
825        let request_2 = MediaRequestParameters {
826            source: MediaSource::Plain(uri_2.to_owned()),
827            format: MediaFormat::File,
828        };
829
830        let now = SystemTime::UNIX_EPOCH;
831
832        let store = MockMediaStoreInner::default();
833        let service = MediaService::with_time_provider(MockTimeProvider::new(now));
834
835        // Set an empty policy.
836        let policy = MediaRetentionPolicy::empty();
837        service.set_media_retention_policy(&store, policy).await.unwrap();
838
839        // Add the contents.
840        service
841            .add_media_content(&store, &request_1, content.clone(), IgnoreMediaRetentionPolicy::No)
842            .await
843            .unwrap();
844        service
845            .add_media_content(&store, &request_2, content, IgnoreMediaRetentionPolicy::No)
846            .await
847            .unwrap();
848        assert!(service.inner.automatic_media_cleanup_join_handle.lock().is_none());
849
850        // Try to launch an automatic cleanup.
851        let now = now + Duration::from_secs(60);
852        service.inner.time_provider.set_now(now);
853        service.maybe_spawn_automatic_media_cache_cleanup(&store, now);
854
855        // No cleanup was spawned since automatic cleanups are disabled.
856        assert!(service.inner.automatic_media_cleanup_join_handle.lock().is_none());
857
858        // Set a policy with automatic cleanup every hour.
859        let policy = MediaRetentionPolicy::empty()
860            .with_cleanup_frequency(Some(Duration::from_secs(60 * 60)));
861        let now = now + Duration::from_secs(60);
862        service.inner.time_provider.set_now(now);
863        service.set_media_retention_policy(&store, policy).await.unwrap();
864
865        // No cleanup was spawned since the policy has no limitations.
866        assert!(service.inner.automatic_media_cleanup_join_handle.lock().is_none());
867
868        // Set a policy with automatic cleanup every hour and a max file size.
869        let policy = MediaRetentionPolicy::empty()
870            .with_cleanup_frequency(Some(Duration::from_secs(60 * 60)))
871            .with_max_file_size(Some(512));
872        let now = now + Duration::from_secs(60);
873        service.inner.time_provider.set_now(now);
874        service.set_media_retention_policy(&store, policy).await.unwrap();
875
876        // A cleanup was spawned since there was no last_media_cleanup_time.
877        let join_handle = service.inner.automatic_media_cleanup_join_handle.lock().take().unwrap();
878        join_handle.await.unwrap();
879
880        assert_eq!(store.last_media_cleanup_time_inner().await.unwrap(), Some(now));
881
882        // Try again one minute in the future, nothing is spawned because we
883        // need to wait for one hour.
884        let now = now + Duration::from_secs(60);
885        service.inner.time_provider.set_now(now);
886        service.get_media_content(&store, &request_1).await.unwrap();
887
888        assert!(service.inner.automatic_media_cleanup_join_handle.lock().is_none());
889
890        // Try again 2 hours in the future, another cleanup is spawned.
891        let now = now + Duration::from_secs(2 * 60 * 60);
892        service.inner.time_provider.set_now(now);
893        service.get_media_content(&store, &request_1).await.unwrap();
894
895        let join_handle = service.inner.automatic_media_cleanup_join_handle.lock().take().unwrap();
896        join_handle.await.unwrap();
897
898        assert_eq!(store.last_media_cleanup_time_inner().await.unwrap(), Some(now));
899    }
900}
matrix_sdk_base/media/store/media_service.rs

matrix_sdk_base/media/store/
media_service.rs
