tower_sessions_ext_core/
session_store.rs

1//! A session backend for managing session state.
2//!
3//! This crate provides the ability to use custom backends for session
4//! management by implementing the [`SessionStore`] trait. This trait defines
5//! the necessary operations for creating, saving, loading, and deleting session
6//! records.
7//!
8//! # Implementing a Custom Store
9//!
10//! Below is an example of implementing a custom session store using an
11//! in-memory [`HashMap`]. This example is for illustration purposes only; you
12//! can use the provided [`MemoryStore`] directly without implementing it
13//! yourself.
14//!
15//! ```rust
16//! use std::{collections::HashMap, sync::Arc};
17//!
18//! use async_trait::async_trait;
19//! use time::OffsetDateTime;
20//! use tokio::sync::Mutex;
21//! use tower_sessions_ext_core::{
22//!     SessionStore,
23//!     session::{Id, Record},
24//!     session_store,
25//! };
26//!
27//! #[derive(Clone, Debug, Default)]
28//! pub struct MemoryStore(Arc<Mutex<HashMap<Id, Record>>>);
29//!
30//! #[async_trait]
31//! impl SessionStore for MemoryStore {
32//!     async fn create(&self, record: &mut Record) -> session_store::Result<()> {
33//!         let mut store_guard = self.0.lock().await;
34//!         while store_guard.contains_key(&record.id) {
35//!             // Session ID collision mitigation.
36//!             record.id = Id::default();
37//!         }
38//!         store_guard.insert(record.id, record.clone());
39//!         Ok(())
40//!     }
41//!
42//!     async fn save(&self, record: &Record) -> session_store::Result<()> {
43//!         self.0.lock().await.insert(record.id, record.clone());
44//!         Ok(())
45//!     }
46//!
47//!     async fn load(&self, session_id: &Id) -> session_store::Result<Option<Record>> {
48//!         Ok(self
49//!             .0
50//!             .lock()
51//!             .await
52//!             .get(session_id)
53//!             .filter(|Record { expiry_date, .. }| is_active(*expiry_date))
54//!             .cloned())
55//!     }
56//!
57//!     async fn delete(&self, session_id: &Id) -> session_store::Result<()> {
58//!         self.0.lock().await.remove(session_id);
59//!         Ok(())
60//!     }
61//! }
62//!
63//! fn is_active(expiry_date: OffsetDateTime) -> bool {
64//!     expiry_date > OffsetDateTime::now_utc()
65//! }
66//! ```
67//!
68//! # Session Store Trait
69//!
70//! The [`SessionStore`] trait defines the interface for session management.
71//! Implementations must handle session creation, saving, loading, and deletion.
72//!
73//! # CachingSessionStore
74//!
75//! The [`CachingSessionStore`] provides a layered caching mechanism with a
76//! cache as the frontend and a store as the backend. This can improve read
77//! performance by reducing the need to access the backend store for frequently
78//! accessed sessions.
79//!
80//! # ExpiredDeletion
81//!
82//! The [`ExpiredDeletion`] trait provides a method for deleting expired
83//! sessions. Implementations can optionally provide a method for continuously
84//! deleting expired sessions at a specified interval.
85use std::fmt::Debug;
86
87use async_trait::async_trait;
88
89use crate::session::{Id, Record};
90
91/// Stores must map any errors that might occur during their use to this type.
92#[derive(thiserror::Error, Debug)]
93pub enum Error {
94    #[error("Encoding failed with: {0}")]
95    Encode(String),
96
97    #[error("Decoding failed with: {0}")]
98    Decode(String),
99
100    #[error("{0}")]
101    Backend(String),
102}
103
104pub type Result<T> = std::result::Result<T, Error>;
105
106/// Defines the interface for session management.
107///
108/// See [`session_store`](crate::session_store) for more details.
109#[async_trait]
110pub trait SessionStore: Debug + Send + Sync + 'static {
111    /// Creates a new session in the store with the provided session record.
112    ///
113    /// Implementers must decide how to handle potential ID collisions. For
114    /// example, they might generate a new unique ID or return `Error::Backend`.
115    ///
116    /// The record is given as an exclusive reference to allow modifications,
117    /// such as assigning a new ID, during the creation process.
118    async fn create(&self, session_record: &mut Record) -> Result<()> {
119        default_create(self, session_record).await
120    }
121
122    /// Saves the provided session record to the store.
123    ///
124    /// This method is intended for updating the state of an existing session.
125    async fn save(&self, session_record: &Record) -> Result<()>;
126
127    /// Loads an existing session record from the store using the provided ID.
128    ///
129    /// If a session with the given ID exists, it is returned. If the session
130    /// does not exist or has been invalidated (e.g., expired), `None` is
131    /// returned.
132    async fn load(&self, session_id: &Id) -> Result<Option<Record>>;
133
134    /// Deletes a session record from the store using the provided ID.
135    ///
136    /// If the session exists, it is removed from the store.
137    async fn delete(&self, session_id: &Id) -> Result<()>;
138}
139
140async fn default_create<S: SessionStore + ?Sized>(
141    store: &S,
142    session_record: &mut Record,
143) -> Result<()> {
144    tracing::warn!(
145        "The default implementation of `SessionStore::create` is being used, which relies on \
146         `SessionStore::save`. To properly handle potential ID collisions, it is recommended that \
147         stores implement their own version of `SessionStore::create`."
148    );
149    store.save(session_record).await?;
150    Ok(())
151}
152
153/// Provides a layered caching mechanism with a cache as the frontend and a
154/// store as the backend..
155///
156/// Contains both a cache, which acts as a frontend, and a store which acts as a
157/// backend. Both cache and store implement `SessionStore`.
158///
159/// By using a cache, the cost of reads can be greatly reduced as once cached,
160/// reads need only interact with the frontend, forgoing the cost of retrieving
161/// the session record from the backend.
162///
163/// # Examples
164///
165/// ```rust,ignore
166/// # tokio_test::block_on(async {
167/// use tower_sessions_ext::CachingSessionStore;
168/// use tower_sessions_ext_moka_store::MokaStore;
169/// use tower_sessions_ext_sqlx_store::{SqlitePool, SqliteStore};
170/// let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
171/// let sqlite_store = SqliteStore::new(pool);
172/// let moka_store = MokaStore::new(Some(2_000));
173/// let caching_store = CachingSessionStore::new(moka_store, sqlite_store);
174/// # })
175/// ```
176#[derive(Debug, Clone)]
177pub struct CachingSessionStore<Cache: SessionStore, Store: SessionStore> {
178    cache: Cache,
179    store: Store,
180}
181
182impl<Cache: SessionStore, Store: SessionStore> CachingSessionStore<Cache, Store> {
183    /// Create a new `CachingSessionStore`.
184    pub fn new(cache: Cache, store: Store) -> Self {
185        Self { cache, store }
186    }
187}
188
189#[async_trait]
190impl<Cache, Store> SessionStore for CachingSessionStore<Cache, Store>
191where
192    Cache: SessionStore,
193    Store: SessionStore,
194{
195    async fn create(&self, record: &mut Record) -> Result<()> {
196        self.store.create(record).await?;
197        self.cache.create(record).await?;
198        Ok(())
199    }
200
201    async fn save(&self, record: &Record) -> Result<()> {
202        let store_save_fut = self.store.save(record);
203        let cache_save_fut = self.cache.save(record);
204
205        futures::try_join!(store_save_fut, cache_save_fut)?;
206
207        Ok(())
208    }
209
210    async fn load(&self, session_id: &Id) -> Result<Option<Record>> {
211        match self.cache.load(session_id).await {
212            // We found a session in the cache, so let's use it.
213            Ok(Some(session_record)) => Ok(Some(session_record)),
214
215            // We didn't find a session in the cache, so we'll try loading from the backend.
216            //
217            // When we find a session in the backend, we'll hydrate our cache with it.
218            Ok(None) => {
219                let session_record = self.store.load(session_id).await?;
220
221                if let Some(ref session_record) = session_record {
222                    self.cache.save(session_record).await?;
223                }
224
225                Ok(session_record)
226            }
227
228            // Some error occurred with our cache so we'll bubble this up.
229            Err(err) => Err(err),
230        }
231    }
232
233    async fn delete(&self, session_id: &Id) -> Result<()> {
234        let store_delete_fut = self.store.delete(session_id);
235        let cache_delete_fut = self.cache.delete(session_id);
236
237        futures::try_join!(store_delete_fut, cache_delete_fut)?;
238
239        Ok(())
240    }
241}
242
243/// Provides a method for deleting expired sessions.
244#[async_trait]
245pub trait ExpiredDeletion: SessionStore
246where
247    Self: Sized,
248{
249    /// Deletes expired sessions from the store, returning their full records.
250    async fn delete_expired(&self) -> Result<Vec<Record>>;
251
252    /// Runs indefinitely, deleting expired sessions on each tick.
253    /// After each deletion round, invokes `on_deleted` with the full
254    /// records of sessions that were just removed.
255    ///
256    /// Generally this will be used as a task, for example via
257    /// `tokio::task::spawn`.
258    ///
259    /// # Errors
260    ///
261    /// This function returns a `Result` that contains an error if the
262    /// deletion operation fails.
263    ///
264    /// # Examples
265    ///
266    /// ```rust,no_run,ignore
267    /// use tower_sessions_ext::session_store::ExpiredDeletion;
268    /// use tower_sessions_ext_sqlx_store::{sqlx::SqlitePool, SqliteStore};
269    ///
270    /// # {
271    /// # tokio_test::block_on(async {
272    /// let pool = SqlitePool::connect("sqlite::memory:").await.unwrap();
273    /// let session_store = SqliteStore::new(pool);
274    ///
275    /// tokio::task::spawn(
276    ///     session_store
277    ///         .clone()
278    ///         .continuously_delete_expired(tokio::time::Duration::from_secs(60), |_records| {
279    ///             Box::pin(async {})
280    ///         }),
281    /// );
282    /// # })
283    /// ```
284    #[cfg(feature = "deletion-task")]
285    #[cfg_attr(docsrs, doc(cfg(feature = "deletion-task")))]
286    async fn continuously_delete_expired(
287        self,
288        period: tokio::time::Duration,
289        on_deleted: impl Fn(Vec<Record>) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync + 'static,
290    ) -> Result<()> {
291        let mut interval = tokio::time::interval(period);
292        interval.tick().await; // The first tick completes immediately; skip.
293        loop {
294            interval.tick().await;
295            let deleted = self.delete_expired().await?;
296            if !deleted.is_empty() {
297                on_deleted(deleted).await;
298            }
299        }
300    }
301}
302
303#[cfg(test)]
304mod tests {
305    use mockall::{
306        mock,
307        predicate::{self, *},
308    };
309    use time::{Duration, OffsetDateTime};
310
311    use super::*;
312
313    mock! {
314        #[derive(Debug)]
315        pub Cache {}
316
317        #[async_trait]
318        impl SessionStore for Cache {
319            async fn create(&self, record: &mut Record) -> Result<()>;
320            async fn save(&self, record: &Record) -> Result<()>;
321            async fn load(&self, session_id: &Id) -> Result<Option<Record>>;
322            async fn delete(&self, session_id: &Id) -> Result<()>;
323        }
324    }
325
326    mock! {
327        #[derive(Debug)]
328        pub Store {}
329
330        #[async_trait]
331        impl SessionStore for Store {
332            async fn create(&self, record: &mut Record) -> Result<()>;
333            async fn save(&self, record: &Record) -> Result<()>;
334            async fn load(&self, session_id: &Id) -> Result<Option<Record>>;
335            async fn delete(&self, session_id: &Id) -> Result<()>;
336        }
337    }
338
339    mock! {
340        #[derive(Debug)]
341        pub CollidingStore {}
342
343        #[async_trait]
344        impl SessionStore for CollidingStore {
345            async fn save(&self, record: &Record) -> Result<()>;
346            async fn load(&self, session_id: &Id) -> Result<Option<Record>>;
347            async fn delete(&self, session_id: &Id) -> Result<()>;
348        }
349    }
350
351    #[tokio::test]
352    async fn test_create() {
353        let mut store = MockCollidingStore::new();
354        let mut record = Record {
355            id: Default::default(),
356            data: Default::default(),
357            expiry_date: OffsetDateTime::now_utc() + Duration::minutes(30),
358        };
359
360        store
361            .expect_save()
362            .with(predicate::eq(record.clone()))
363            .times(1)
364            .returning(|_| Ok(()));
365        let result = store.create(&mut record).await;
366        assert!(result.is_ok());
367    }
368
369    #[tokio::test]
370    async fn test_save() {
371        let mut store = MockStore::new();
372        let record = Record {
373            id: Default::default(),
374            data: Default::default(),
375            expiry_date: OffsetDateTime::now_utc() + Duration::minutes(30),
376        };
377        store
378            .expect_save()
379            .with(predicate::eq(record.clone()))
380            .times(1)
381            .returning(|_| Ok(()));
382
383        let result = store.save(&record).await;
384        assert!(result.is_ok());
385    }
386
387    #[tokio::test]
388    async fn test_load() {
389        let mut store = MockStore::new();
390        let session_id = Id::default();
391        let record = Record {
392            id: Default::default(),
393            data: Default::default(),
394            expiry_date: OffsetDateTime::now_utc() + Duration::minutes(30),
395        };
396        let expected_record = record.clone();
397
398        store
399            .expect_load()
400            .with(predicate::eq(session_id))
401            .times(1)
402            .returning(move |_| Ok(Some(record.clone())));
403
404        let result = store.load(&session_id).await;
405        assert!(result.is_ok());
406        assert_eq!(result.unwrap(), Some(expected_record));
407    }
408
409    #[tokio::test]
410    async fn test_delete() {
411        let mut store = MockStore::new();
412        let session_id = Id::default();
413
414        store
415            .expect_delete()
416            .with(predicate::eq(session_id))
417            .times(1)
418            .returning(|_| Ok(()));
419
420        let result = store.delete(&session_id).await;
421        assert!(result.is_ok());
422    }
423
424    #[tokio::test]
425    async fn test_caching_store_create() {
426        let mut cache = MockCache::new();
427        let mut store = MockStore::new();
428        let mut record = Record {
429            id: Default::default(),
430            data: Default::default(),
431            expiry_date: OffsetDateTime::now_utc() + Duration::minutes(30),
432        };
433
434        cache.expect_create().times(1).returning(|_| Ok(()));
435        store.expect_create().times(1).returning(|_| Ok(()));
436
437        let caching_store = CachingSessionStore::new(cache, store);
438        let result = caching_store.create(&mut record).await;
439        assert!(result.is_ok());
440    }
441
442    #[tokio::test]
443    async fn test_caching_store_save() {
444        let mut cache = MockCache::new();
445        let mut store = MockStore::new();
446        let record = Record {
447            id: Default::default(),
448            data: Default::default(),
449            expiry_date: OffsetDateTime::now_utc() + Duration::minutes(30),
450        };
451
452        cache
453            .expect_save()
454            .with(predicate::eq(record.clone()))
455            .times(1)
456            .returning(|_| Ok(()));
457        store
458            .expect_save()
459            .with(predicate::eq(record.clone()))
460            .times(1)
461            .returning(|_| Ok(()));
462
463        let caching_store = CachingSessionStore::new(cache, store);
464        let result = caching_store.save(&record).await;
465        assert!(result.is_ok());
466    }
467
468    #[tokio::test]
469    async fn test_caching_store_load() {
470        let mut cache = MockCache::new();
471        let mut store = MockStore::new();
472        let session_id = Id::default();
473        let record = Record {
474            id: Default::default(),
475            data: Default::default(),
476            expiry_date: OffsetDateTime::now_utc() + Duration::minutes(30),
477        };
478        let expected_record = record.clone();
479
480        cache
481            .expect_load()
482            .with(predicate::eq(session_id))
483            .times(1)
484            .returning(move |_| Ok(Some(record.clone())));
485        // Store load should not be called since cache returns a record
486        store.expect_load().times(0);
487
488        let caching_store = CachingSessionStore::new(cache, store);
489        let result = caching_store.load(&session_id).await;
490        assert!(result.is_ok());
491        assert_eq!(result.unwrap(), Some(expected_record));
492    }
493
494    #[tokio::test]
495    async fn test_caching_store_delete() {
496        let mut cache = MockCache::new();
497        let mut store = MockStore::new();
498        let session_id = Id::default();
499
500        cache
501            .expect_delete()
502            .with(predicate::eq(session_id))
503            .times(1)
504            .returning(|_| Ok(()));
505        store
506            .expect_delete()
507            .with(predicate::eq(session_id))
508            .times(1)
509            .returning(|_| Ok(()));
510
511        let caching_store = CachingSessionStore::new(cache, store);
512        let result = caching_store.delete(&session_id).await;
513        assert!(result.is_ok());
514    }
515}
tower_sessions_ext_core/session_store.rs

tower_sessions_ext_core/
session_store.rs
