use std::{convert::TryInto, sync::Arc};
use actix_web::cookie::time::Duration;
use anyhow::{Context, Error};
use redis::{aio::ConnectionManager, AsyncCommands, Cmd, FromRedisValue, RedisResult, Value};
use super::SessionKey;
use crate::storage::{
interface::{LoadError, SaveError, SessionState, UpdateError},
utils::generate_session_key,
SessionStore,
};
#[derive(Clone)]
pub struct RedisSessionStore {
configuration: CacheConfiguration,
client: ConnectionManager,
}
#[derive(Clone)]
struct CacheConfiguration {
cache_keygen: Arc<dyn Fn(&str) -> String + Send + Sync>,
}
impl Default for CacheConfiguration {
fn default() -> Self {
Self {
cache_keygen: Arc::new(str::to_owned),
}
}
}
impl RedisSessionStore {
pub fn builder<S: Into<String>>(connection_string: S) -> RedisSessionStoreBuilder {
RedisSessionStoreBuilder {
configuration: CacheConfiguration::default(),
connection_string: connection_string.into(),
}
}
pub async fn new<S: Into<String>>(
connection_string: S,
) -> Result<RedisSessionStore, anyhow::Error> {
Self::builder(connection_string).build().await
}
}
#[must_use]
pub struct RedisSessionStoreBuilder {
connection_string: String,
configuration: CacheConfiguration,
}
impl RedisSessionStoreBuilder {
pub fn cache_keygen<F>(mut self, keygen: F) -> Self
where
F: Fn(&str) -> String + 'static + Send + Sync,
{
self.configuration.cache_keygen = Arc::new(keygen);
self
}
pub async fn build(self) -> Result<RedisSessionStore, anyhow::Error> {
let client = ConnectionManager::new(redis::Client::open(self.connection_string)?).await?;
Ok(RedisSessionStore {
configuration: self.configuration,
client,
})
}
}
#[async_trait::async_trait(?Send)]
impl SessionStore for RedisSessionStore {
async fn load(&self, session_key: &SessionKey) -> Result<Option<SessionState>, LoadError> {
let cache_key = (self.configuration.cache_keygen)(session_key.as_ref());
let value: Option<String> = self
.execute_command(redis::cmd("GET").arg(&[&cache_key]))
.await
.map_err(Into::into)
.map_err(LoadError::Other)?;
match value {
None => Ok(None),
Some(value) => Ok(serde_json::from_str(&value)
.map_err(Into::into)
.map_err(LoadError::Deserialization)?),
}
}
async fn save(
&self,
session_state: SessionState,
ttl: &Duration,
) -> Result<SessionKey, SaveError> {
let user_id = session_state.get("user_id");
let body = serde_json::to_string(&session_state)
.map_err(Into::into)
.map_err(SaveError::Serialization)?;
let session_key = generate_session_key(user_id.and_then(|value| Some(value.to_string())));
let cache_key = (self.configuration.cache_keygen)(session_key.as_ref());
self.execute_command(redis::cmd("SET").arg(&[
&cache_key,
&body,
"NX", "EX", &format!("{}", ttl.whole_seconds()),
]))
.await
.map_err(Into::into)
.map_err(SaveError::Other)?;
Ok(session_key)
}
async fn update(
&self,
session_key: SessionKey,
session_state: SessionState,
ttl: &Duration,
) -> Result<SessionKey, UpdateError> {
let body = serde_json::to_string(&session_state)
.map_err(Into::into)
.map_err(UpdateError::Serialization)?;
let cache_key = (self.configuration.cache_keygen)(session_key.as_ref());
let v: redis::Value = self
.execute_command(redis::cmd("SET").arg(&[
&cache_key,
&body,
"XX", "EX", &format!("{}", ttl.whole_seconds()),
]))
.await
.map_err(Into::into)
.map_err(UpdateError::Other)?;
match v {
Value::Nil => {
self.save(session_state, ttl)
.await
.map_err(|err| match err {
SaveError::Serialization(err) => UpdateError::Serialization(err),
SaveError::Other(err) => UpdateError::Other(err),
})
}
Value::Int(_) | Value::Okay | Value::Status(_) => Ok(session_key),
val => Err(UpdateError::Other(anyhow::anyhow!(
"Failed to update session state. {:?}",
val
))),
}
}
async fn update_ttl(&self, session_key: &SessionKey, ttl: &Duration) -> Result<(), Error> {
let cache_key = (self.configuration.cache_keygen)(session_key.as_ref());
self.client
.clone()
.expire(
&cache_key,
ttl.whole_seconds().try_into().context(
"Failed to convert the state TTL into the number of whole seconds remaining",
)?,
)
.await?;
Ok(())
}
async fn delete(&self, session_key: &SessionKey) -> Result<(), anyhow::Error> {
let cache_key = (self.configuration.cache_keygen)(session_key.as_ref());
self.execute_command(redis::cmd("DEL").arg(&[&cache_key]))
.await
.map_err(Into::into)
.map_err(UpdateError::Other)?;
Ok(())
}
}
impl RedisSessionStore {
#[allow(clippy::needless_pass_by_ref_mut)]
async fn execute_command<T: FromRedisValue>(&self, cmd: &mut Cmd) -> RedisResult<T> {
let mut can_retry = true;
loop {
match cmd.query_async(&mut self.client.clone()).await {
Ok(value) => return Ok(value),
Err(err) => {
if can_retry && err.is_connection_dropped() {
tracing::debug!(
"Connection dropped while trying to talk to Redis. Retrying."
);
can_retry = false;
continue;
} else {
return Err(err);
}
}
}
}
}
}
#[cfg(test)]
mod tests {
use actix_web::cookie::time;
use redis::AsyncCommands;
use serde_json::Map;
use super::*;
use crate::test_helpers::acceptance_test_suite;
async fn redis_store() -> RedisSessionStore {
RedisSessionStore::new("redis://127.0.0.1:7001")
.await
.unwrap()
}
#[actix_web::test]
async fn test_session_workflow() {
let redis_store = redis_store().await;
acceptance_test_suite(move || redis_store.clone(), true).await;
}
#[actix_web::test]
async fn loading_a_missing_session_returns_none() {
let store = redis_store().await;
let session_key = generate_session_key(None);
assert!(store.load(&session_key).await.unwrap().is_none());
}
#[actix_web::test]
async fn loading_an_invalid_session_state_returns_deserialization_error() {
let store = redis_store().await;
let session_key = generate_session_key(None);
store
.client
.clone()
.set::<_, _, ()>(session_key.as_ref(), "random-thing-which-is-not-json")
.await
.unwrap();
assert!(matches!(
store.load(&session_key).await.unwrap_err(),
LoadError::Deserialization(_),
));
}
#[actix_web::test]
async fn updating_of_an_expired_state_is_handled_gracefully() {
let store = redis_store().await;
let session_key = generate_session_key(None);
let initial_session_key = session_key.as_ref().to_owned();
let updated_session_key = store
.update(session_key, Map::new(), &time::Duration::seconds(1))
.await
.unwrap();
assert_ne!(initial_session_key, updated_session_key.as_ref());
}
}