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

matrix_sdk_base/media/store/
media_service.rs
