Merge pull request #40 from storytold/bt-cache-session

cache sessions + purge cache on billing, edit profile, etc.
This commit is contained in:
Brandon Thomas authored and GitHub committed 2023-03-12 05:19:28 -04:00
commit 495c01c38c
23 files changed
+399 -33

No files matched your search

Generated
+13
View File
@@ -2101,6 +2101,7 @@ dependencies = [
"serde_derive",
"serde_json",
"sqlx",
"sqlx_mysql_helpers",
"tokens",
"toml",
]
@@ -6322,6 +6323,14 @@ dependencies = [
"tokio-native-tls",
]
[[package]]
name = "sqlx_mysql_helpers"
version = "0.0.1"
dependencies = [
"errors",
"sqlx",
]
[[package]]
name = "stable_deref_trait"
version = "1.2.0"
@@ -7750,6 +7759,7 @@ dependencies = [
"database_queries",
"dotenv",
"enums",
"errors",
"hmac 0.10.1",
"http_server_common",
"jwt",
@@ -7758,6 +7768,8 @@ dependencies = [
"md-5",
"once_cell",
"r2d2_redis",
"redis_caching",
"redis_common",
"regex",
"reusable_types",
"serde",
@@ -7765,6 +7777,7 @@ dependencies = [
"serde_json",
"sha2 0.9.5",
"sqlx",
"sqlx_mysql_helpers",
"time 0.2.26",
"tokens",
"toml",
+1
View File
@@ -36,6 +36,7 @@ members = [
"crates/lib/media",
"crates/lib/redis_common",
"crates/lib/sqlite_queries",
"crates/lib/sqlx_mysql_helpers",
"crates/lib/subprocess_common",
"crates/lib/tts_common",
"crates/lib/url_config",
+1
View File
@@ -21,6 +21,7 @@ container_common = { path = "../container_common" }
enums = { path = "../../_types/enums" }
errors = { path = "../errors" }
reusable_types = { path = "../../_types/reusable_types" }
sqlx_mysql_helpers = { path = "../sqlx_mysql_helpers" }
tokens = { path = "../../_types/tokens" }
# External
@@ -7,10 +7,11 @@ use errors::AnyhowResult;
use crate::helpers::boolean_converters::{nullable_i8_to_optional_bool, i8_to_bool, nullable_i8_to_bool_default_false};
use enums::common::visibility::Visibility;
use log::warn;
use sqlx::MySql;
use sqlx::{Executor, MySql};
use sqlx::pool::PoolConnection;
use tokens::users::user::UserToken;
#[derive(Clone, Serialize, Deserialize)]
pub struct SessionUserRecord {
pub user_token: String, // TODO(bt, 2022-12-20): Convert to strongly-typed `UserToken`
pub user_token_typed: UserToken,
@@ -74,11 +75,20 @@ impl SessionUserRecord {
}
}
pub async fn get_user_session_by_token(
pub async fn get_user_session_by_token_pooled_connection(
mysql_connection: &mut PoolConnection<MySql>,
session_token: &str,
) -> AnyhowResult<Option<SessionUserRecord>> {
get_user_session_by_token(mysql_connection, session_token).await
}
pub async fn get_user_session_by_token<'e, 'c : 'e, E>(
mysql_executor: E,
session_token: &str,
) -> AnyhowResult<Option<SessionUserRecord>>
where E: 'e + Executor<'c, Database = MySql>
{
// NB: Lookup failure is Err(RowNotFound).
let maybe_user_record = sqlx::query_as!(
SessionUserRawDbRecord,
@@ -138,7 +148,7 @@ WHERE user_sessions.token = ?
"#,
session_token,
)
.fetch_one(mysql_connection)
.fetch_one(mysql_executor)
.await; // TODO: This will return error if it doesn't exist
match maybe_user_record {
@@ -5,19 +5,20 @@
use anyhow::anyhow;
use errors::AnyhowResult;
use log::warn;
use sqlx::MySql;
use sqlx::pool::PoolConnection;
use sqlx::{Executor, MySql};
#[derive(Clone, Serialize, Deserialize)]
pub struct SessionRecord {
pub session_token: String,
pub user_token: String,
}
pub async fn get_user_session_by_token_light(
mysql_connection: &mut PoolConnection<MySql>,
session_token: &str,
) -> AnyhowResult<Option<SessionRecord>> {
// NB: Lookup failure is Err(RowNotFound).
pub async fn get_user_session_by_token_light<'e, 'c : 'e, E>(
mysql_executor: E,
session_token: &str,
) -> AnyhowResult<Option<SessionRecord>>
where E: 'e + Executor<'c, Database = MySql>
{
let maybe_session_record = sqlx::query_as!(
SessionRecord,
r#"
@@ -30,7 +31,7 @@ AND deleted_at IS NULL
"#,
session_token,
)
.fetch_one(mysql_connection)
.fetch_one(mysql_executor)
.await;
match maybe_session_record {
@@ -8,4 +8,12 @@ impl RedisCacheKeys {
pub fn get_tts_model_endpoint(model_token: &str) -> String {
format!("get_tts_model:{}", model_token)
}
pub fn session_record_light(session_token: &str) -> String {
format!("ses_light:{}", session_token)
}
pub fn session_record_user(session_token: &str) -> String {
format!("ses_user:{}", session_token)
}
}
+26
View File
@@ -0,0 +1,26 @@
[package]
name = "sqlx_mysql_helpers"
edition = "2018"
version = "0.0.1"
authors = [
"Brandon Thomas <bt@brand.io>",
"Brandon Thomas <echelon@gmail.com>",
]
publish = false
[lib]
name = "sqlx_mysql_helpers"
path = "src/lib.rs"
[dependencies]
# Internal
errors = { path = "../errors" }
# MySql Database
# "offline" supports typesafe builds without a live database
# "rustls" doesn't appear to work with IP addresses, per:
# https://github.com/launchbadge/sqlx/issues/1095#issuecomment-798050828
sqlx = { version = "0.6.2", features = [ "mysql", "runtime-actix-native-tls", "offline", "chrono" ] }
[dev-dependencies]
# None yet
@@ -0,0 +1,78 @@
use sqlx::{MySql, MySqlPool};
use sqlx::pool::PoolConnection;
use errors::AnyhowResult;
/// Pass a handle to MySql without caring about the type.
pub enum MySqlHandle {
ConnectionPool(MySqlPool),
Connection(PoolConnection<MySql>),
}
/// Pass a handle to MySql without caring about the type.
/// NB: The connection is typically owned, whereas the connection pool is not.
pub enum MySqlHandleMixedRef<'a> {
ConnectionPool(&'a MySqlPool),
Connection(PoolConnection<MySql>),
}
/// Pass a handle to MySql without caring about the type.
pub enum MySqlHandleRef<'a> {
ConnectionPool(&'a MySqlPool),
Connection(&'a mut PoolConnection<MySql>),
}
//impl MySqlHandle {
// pub fn as_ref(&mut self) -> MySqlHandleRef {
// match self {
// Self::ConnectionPool(inner) => MySqlHandleRef::ConnectionPool(inner),
// Self::Connection(inner) => MySqlHandleRef::Connection(inner),
// }
// }
//}
impl <'a> MySqlHandleMixedRef<'a> {
/// Consume the handle and return a connection.
pub async fn to_discrete_connection(self) -> AnyhowResult<PoolConnection<MySql>> {
match self {
Self::Connection(connection) => Ok(connection),
Self::ConnectionPool(pool) => Ok(pool.acquire().await?),
}
}
}
//impl <'a> MySqlHandleRef<'a> {
// /// Consume the handle and return a connection.
// pub async fn to_discrete_connection(self) -> AnyhowResult<()> {
// let _r = match self {
// Self::Connection(connection) => Ok(connection),
// Self::ConnectionPool(pool) => {
// let connection = pool.acquire().await?;
// Ok(connection)
// },
// };
//
// Ok(())
// }
//}
//
///// Exists to hold ownership (if necessary)
//pub struct MySqlConnectionWrapper<'a> {
// maybe_owned: Option<PoolConnection<MySql>>,
// reference: &'a mut PoolConnection<MySql>,
//}
// TODO:
//impl <'a> MySqlConnectionRef<'a> {
// pub async fn get_pool_connection(&'a mut self) -> AnyhowResult<&'a PoolConnection<MySql>> {
// match self {
// MySqlConnectionRef::ConnectionPool(pool) => {
// let connection = pool.acquire().await?;
// Ok(connection)
// }
// MySqlConnectionRef::Connection(inner) => Ok(inner),
// }
//
// }
//}
+20
View File
@@ -0,0 +1,20 @@
//! sqlx_mysql_helpers
//!
//! Collect reusable Sqlx functions and patterns, particularly for MySQL databases.
//!
// Never allow these
#![forbid(private_in_public)]
#![forbid(unused_must_use)] // NB: It's unsafe to not close/check some things
// Okay to toggle
#![forbid(unreachable_patterns)]
#![forbid(unused_imports)]
#![forbid(unused_mut)]
#![forbid(unused_variables)]
// Always allow
#![allow(dead_code)]
#![allow(non_snake_case)]
pub mod any_connection;
@@ -8,15 +8,16 @@ use crate::stripe::http_endpoints::checkout::create::stripe_create_checkout_sess
use crate::stripe::http_endpoints::checkout::create::stripe_create_checkout_session_shared::stripe_create_checkout_session_shared;
use crate::stripe::stripe_config::StripeConfig;
use crate::stripe::traits::internal_product_to_stripe_lookup::InternalProductToStripeLookup;
use crate::stripe::traits::internal_session_cache_purge::InternalSessionCachePurge;
use crate::stripe::traits::internal_user_lookup::InternalUserLookup;
use http_server_common::request::get_request_header_optional::get_request_header_optional;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
use log::{error, warn};
use reusable_types::server_environment::ServerEnvironment;
use sqlx::MySqlPool;
use std::collections::HashMap;
use std::fmt;
use stripe::{CheckoutSession, CheckoutSessionMode, CreateCheckoutSession, CreateCheckoutSessionLineItems};
use reusable_types::server_environment::ServerEnvironment;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
// =============== Request ===============
@@ -45,6 +46,7 @@ pub async fn stripe_create_checkout_session_json_handler(
url_redirector: web::Data<ThirdPartyUrlRedirector>,
internal_product_to_stripe_lookup: web::Data<dyn InternalProductToStripeLookup>,
internal_user_lookup: web::Data<dyn InternalUserLookup>,
internal_session_cache_purge: web::Data<dyn InternalSessionCachePurge>,
) -> Result<HttpResponse, CreateCheckoutSessionError>
{
let maybe_internal_product_key = request.internal_plan_key.as_deref();
@@ -60,6 +62,9 @@ pub async fn stripe_create_checkout_session_json_handler(
internal_user_lookup.get_ref(),
).await?;
// Best effort to delete Redis session cache
internal_session_cache_purge.best_effort_purge_session_cache(&http_request);
let response = CreateCheckoutSessionSuccessResponse {
success: true,
stripe_checkout_redirect_url: url,
@@ -8,14 +8,15 @@ use crate::stripe::http_endpoints::checkout::create::stripe_create_checkout_sess
use crate::stripe::http_endpoints::checkout::create::stripe_create_checkout_session_shared::stripe_create_checkout_session_shared;
use crate::stripe::stripe_config::StripeConfig;
use crate::stripe::traits::internal_product_to_stripe_lookup::InternalProductToStripeLookup;
use crate::stripe::traits::internal_session_cache_purge::InternalSessionCachePurge;
use crate::stripe::traits::internal_user_lookup::InternalUserLookup;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
use log::{error, warn};
use reusable_types::server_environment::ServerEnvironment;
use sqlx::MySqlPool;
use std::collections::HashMap;
use std::fmt;
use stripe::{CheckoutSession, CheckoutSessionMode, CreateCheckoutSession, CreateCheckoutSessionLineItems};
use reusable_types::server_environment::ServerEnvironment;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
// =============== Request ===============
@@ -34,6 +35,7 @@ pub async fn stripe_create_checkout_session_redirect_handler(
url_redirector: web::Data<ThirdPartyUrlRedirector>,
internal_product_to_stripe_lookup: web::Data<dyn InternalProductToStripeLookup>,
internal_user_lookup: web::Data<dyn InternalUserLookup>,
internal_session_cache_purge: web::Data<dyn InternalSessionCachePurge>,
) -> Result<HttpResponse, CreateCheckoutSessionError>
{
let maybe_internal_product_key = request.product.as_deref();
@@ -49,6 +51,9 @@ pub async fn stripe_create_checkout_session_redirect_handler(
internal_user_lookup.get_ref(),
).await?;
// Best effort to delete Redis session cache
internal_session_cache_purge.best_effort_purge_session_cache(&http_request);
Ok(HttpResponse::Found()
.append_header((header::LOCATION, url))
.finish())
@@ -4,6 +4,7 @@ use crate::stripe::http_endpoints::customer_portal::stripe_create_customer_porta
use crate::stripe::http_endpoints::customer_portal::stripe_create_customer_portal_session_shared::stripe_create_customer_portal_session_shared;
use crate::stripe::stripe_config::StripeConfig;
use crate::stripe::traits::internal_product_to_stripe_lookup::InternalProductToStripeLookup;
use crate::stripe::traits::internal_session_cache_purge::InternalSessionCachePurge;
use crate::stripe::traits::internal_user_lookup::InternalUserLookup;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
@@ -33,6 +34,7 @@ pub async fn stripe_create_customer_portal_session_json_handler(
url_redirector: web::Data<ThirdPartyUrlRedirector>,
internal_product_to_stripe_lookup: web::Data<dyn InternalProductToStripeLookup>,
internal_user_lookup: web::Data<dyn InternalUserLookup>,
internal_session_cache_purge: web::Data<dyn InternalSessionCachePurge>,
) -> Result<HttpResponse, CreateCustomerPortalSessionError>
{
let portal_config_id = request.portal_config_id
@@ -40,6 +42,9 @@ pub async fn stripe_create_customer_portal_session_json_handler(
.unwrap_or(&stripe_config.portal.default_portal_config_id)
.to_string();
// Best effort to delete Redis session cache
internal_session_cache_purge.best_effort_purge_session_cache(&http_request);
let redirect_url = stripe_create_customer_portal_session_shared(
http_request,
stripe_config,
@@ -1,10 +1,11 @@
use actix_http::header;
use actix_web::{HttpRequest, HttpResponse, web};
use actix_web::web::Query;
use actix_web::{HttpRequest, HttpResponse, web};
use crate::stripe::http_endpoints::customer_portal::stripe_create_customer_portal_session_error::CreateCustomerPortalSessionError;
use crate::stripe::http_endpoints::customer_portal::stripe_create_customer_portal_session_shared::stripe_create_customer_portal_session_shared;
use crate::stripe::stripe_config::StripeConfig;
use crate::stripe::traits::internal_product_to_stripe_lookup::InternalProductToStripeLookup;
use crate::stripe::traits::internal_session_cache_purge::InternalSessionCachePurge;
use crate::stripe::traits::internal_user_lookup::InternalUserLookup;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
@@ -26,6 +27,7 @@ pub async fn stripe_create_customer_portal_session_redirect_handler(
url_redirector: web::Data<ThirdPartyUrlRedirector>,
internal_product_to_stripe_lookup: web::Data<dyn InternalProductToStripeLookup>,
internal_user_lookup: web::Data<dyn InternalUserLookup>,
internal_session_cache_purge: web::Data<dyn InternalSessionCachePurge>,
) -> Result<HttpResponse, CreateCustomerPortalSessionError>
{
let portal_config_id = query.portal_config_id
@@ -33,6 +35,9 @@ pub async fn stripe_create_customer_portal_session_redirect_handler(
.unwrap_or(&stripe_config.portal.default_portal_config_id)
.to_string();
// Best effort to delete Redis session cache
internal_session_cache_purge.best_effort_purge_session_cache(&http_request);
let redirect_url = stripe_create_customer_portal_session_shared(
http_request,
stripe_config,
@@ -0,0 +1,9 @@
use actix_web::HttpRequest;
/// Allows us to purge session caches that might keep stale premium plan/billing data around
#[cfg_attr(test, automock)]
pub trait InternalSessionCachePurge {
/// Best effort attempt to delete
/// Fails silently.
fn best_effort_purge_session_cache(&self, http_request: &HttpRequest);
}
@@ -1,3 +1,4 @@
pub mod internal_product_to_stripe_lookup;
pub mod internal_session_cache_purge;
pub mod internal_subscription_product_lookup;
pub mod internal_user_lookup;
+4
View File
@@ -18,8 +18,12 @@ path = "src/lib.rs"
container_common = { path = "../../../lib/container_common" }
database_queries = { path = "../../../lib/database_queries" }
enums = { path = "../../../_types/enums" }
errors = { path = "../../../lib/errors" }
http_server_common = { path = "../../../lib/http_server_common" }
redis_caching = { path = "../../../lib/caching/redis_caching" }
redis_common = { path = "../../../lib/redis_common" }
reusable_types = { path = "../../../_types/reusable_types" }
sqlx_mysql_helpers = { path = "../../../lib/sqlx_mysql_helpers" }
tokens = { path = "../../../_types/tokens" }
user_input_common = { path = "../../../lib/user_input_common" }
@@ -22,6 +22,8 @@ use enums::common::visibility::Visibility;
use http_server_common::request::get_request_ip::get_request_ip;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
use log::warn;
use redis_caching::redis_ttl_cache::RedisTtlCache;
use redis_common::redis_cache_keys::RedisCacheKeys;
use sqlx::MySqlPool;
use std::fmt;
use user_input_common::check_for_slurs::contains_slurs;
@@ -91,6 +93,7 @@ pub async fn edit_profile_handler(
path: Path<EditProfilePathInfo>,
request: web::Json<EditProfileRequest>,
mysql_pool: web::Data<MySqlPool>,
redis_ttl_cache: web::Data<RedisTtlCache>,
session_checker: web::Data<SessionChecker>,
) -> Result<HttpResponse, EditProfileError>
{
@@ -292,6 +295,22 @@ pub async fn edit_profile_handler(
}
};
// TODO: Clear Redis cache of sessions
// Unfortunately we don't yet have an index of user_token => session_tokens[] outside the DB.
// For now, a hacky solution is just to delete the cache under the current user.
// This makes sense for non-mods and should solve 95% of cases.
if let Some(session_token) = session_checker.forgiving_get_session_token(&http_request) {
if let Ok(mut redis_ttl_cache) = redis_ttl_cache.get_connection() {
let keys = vec![
RedisCacheKeys::session_record_user(&session_token),
RedisCacheKeys::session_record_light(&session_token),
];
for key in keys.iter() {
let _r = redis_ttl_cache.delete_from_cache(key).ok();
}
}
}
let response = EditProfileSuccessResponse {
success: true,
};
@@ -4,18 +4,22 @@
#![forbid(unused_variables)]
use actix_web::HttpRequest;
use container_common::anyhow_result::AnyhowResult;
use crate::utils::session_cookie_manager::SessionCookieManager;
use crate::utils::user_session_extended::{UserSessionExtended, UserSessionPreferences, UserSessionPremiumPlanInfo, UserSessionRoleAndPermissions, UserSessionSubscriptionPlan, UserSessionUserDetails};
use database_queries::queries::users::user_sessions::get_user_session_by_token::{get_user_session_by_token, SessionUserRecord};
use database_queries::queries::users::user_sessions::get_user_session_by_token::{get_user_session_by_token, get_user_session_by_token_pooled_connection, SessionUserRecord};
use database_queries::queries::users::user_sessions::get_user_session_by_token_light::{get_user_session_by_token_light, SessionRecord};
use database_queries::queries::users::user_subscriptions::list_active_user_subscriptions::list_active_user_subscriptions;
use errors::AnyhowResult;
use log::warn;
use redis_caching::redis_ttl_cache::{RedisTtlCache, RedisTtlCacheConnection};
use redis_common::redis_cache_keys::RedisCacheKeys;
use sqlx::pool::PoolConnection;
use sqlx::{MySqlPool, MySql};
use sqlx::{MySqlPool, MySql, Executor};
#[derive(Clone)]
pub struct SessionChecker {
cookie_manager: SessionCookieManager,
maybe_redis_ttl_cache: Option<RedisTtlCache>,
}
impl SessionChecker {
@@ -23,34 +27,87 @@ impl SessionChecker {
pub fn new(cookie_manager: &SessionCookieManager) -> Self {
Self {
cookie_manager: cookie_manager.clone(),
maybe_redis_ttl_cache: None,
}
}
#[deprecated = "Use the PoolConnection<MySql> method instead of the MySqlPool one."]
pub fn new_with_cache(cookie_manager: &SessionCookieManager, redis_ttl_cache: RedisTtlCache) -> Self {
Self {
cookie_manager: cookie_manager.clone(),
maybe_redis_ttl_cache: Some(redis_ttl_cache),
}
}
pub fn get_session_token(&self, request: &HttpRequest) -> AnyhowResult<Option<String>> {
self.cookie_manager.decode_session_token_from_request(request)
}
pub fn forgiving_get_session_token(&self, request: &HttpRequest) -> Option<String> {
self.get_session_token(request).ok().flatten()
}
// ==================== SessionRecord ====================
//#[deprecated = "Use the PoolConnection<MySql> method instead of the MySqlPool one."]
pub async fn maybe_get_session_light(
&self,
request: &HttpRequest,
pool: &MySqlPool
) -> AnyhowResult<Option<SessionRecord>>
{
let mut connection = pool.acquire().await?;
self.maybe_get_session_light_from_connection(request, &mut connection).await
self.do_session_light_lookup_and_cookie_decode(request, pool).await
}
pub async fn maybe_get_session_light_from_connection(
&self,
request: &HttpRequest,
mysql_connection: &mut PoolConnection<MySql>,
) -> AnyhowResult<Option<SessionRecord>>
{
self.do_session_light_lookup_and_cookie_decode(request, mysql_connection).await
}
async fn do_session_light_lookup_and_cookie_decode<'e, 'c : 'e, E>(
&self,
request: &HttpRequest,
mysql_executor: E,
) -> AnyhowResult<Option<SessionRecord>>
where E: 'e + Executor<'c, Database = MySql>
{
let session_token = match self.cookie_manager.decode_session_token_from_request(request)? {
None => return Ok(None),
Some(session_token) => session_token,
};
get_user_session_by_token_light(mysql_connection, &session_token).await
self.do_session_light_lookup(mysql_executor, &session_token).await
}
async fn do_session_light_lookup<'e, 'c : 'e, E>(
&self,
mysql_executor: E,
session_token: &str,
) -> AnyhowResult<Option<SessionRecord>>
where E: 'e + Executor<'c, Database = MySql>
{
match self.maybe_get_redis_cache_connection() {
None => {
get_user_session_by_token_light(mysql_executor, session_token).await
}
Some(mut redis_ttl_cache) => {
let cache_key = RedisCacheKeys::session_record_light(session_token);
redis_ttl_cache.lazy_load_if_not_cached(&cache_key, move || {
get_user_session_by_token_light(mysql_executor, session_token)
}).await
}
}
}
// ==================== SessionUserRecord ====================
//#[deprecated = "Use the PoolConnection<MySql> method instead of the MySqlPool one."]
pub async fn maybe_get_user_session(
&self,
@@ -58,24 +115,60 @@ impl SessionChecker {
pool: &MySqlPool,
) -> AnyhowResult<Option<SessionUserRecord>>
{
let mut connection = pool.acquire().await?;
self.maybe_get_user_session_from_connection(request, &mut connection).await
self.do_user_session_lookup_and_cookie_decode(request, pool).await
}
pub async fn maybe_get_user_session_from_connection(
&self,
request: &HttpRequest,
mysql_connection: &mut PoolConnection<MySql>,
) -> AnyhowResult<Option<SessionUserRecord>>
{
self.do_user_session_lookup_and_cookie_decode(request, mysql_connection).await
}
async fn do_user_session_lookup_and_cookie_decode<'e, 'c : 'e, E>(
&self,
request: &HttpRequest,
mysql_executor: E,
) -> AnyhowResult<Option<SessionUserRecord>>
where E: 'e + Executor<'c, Database = MySql>
{
let session_token = match self.cookie_manager.decode_session_token_from_request(request)? {
None => return Ok(None),
Some(session_token) => session_token,
};
get_user_session_by_token(mysql_connection, &session_token).await
self.do_user_session_lookup(mysql_executor, &session_token).await
}
async fn do_user_session_lookup<'e, 'c : 'e, E>(
&self,
mysql_executor: E,
session_token: &str,
) -> AnyhowResult<Option<SessionUserRecord>>
where E: 'e + Executor<'c, Database = MySql>
{
match self.maybe_get_redis_cache_connection() {
None => {
get_user_session_by_token(mysql_executor, session_token).await
}
Some(mut redis_ttl_cache) => {
let cache_key = RedisCacheKeys::session_record_user(session_token);
redis_ttl_cache.lazy_load_if_not_cached(&cache_key, move || {
get_user_session_by_token(mysql_executor, session_token)
}).await
}
}
}
// ==================== UserSessionExtended ====================
//#[deprecated = "Use the PoolConnection<MySql> method instead of the MySqlPool one."]
pub async fn maybe_get_user_session_extended(
&self,
@@ -100,7 +193,7 @@ impl SessionChecker {
// TODO: Fire both requests off simultaneously.
let user_session = {
match get_user_session_by_token(mysql_connection, &session_payload.session_token).await? {
match get_user_session_by_token_pooled_connection(mysql_connection, &session_payload.session_token).await? {
None => return Ok(None),
Some(u) => u,
}
@@ -164,5 +257,17 @@ impl SessionChecker {
},
}))
}
fn maybe_get_redis_cache_connection(&self) -> Option<RedisTtlCacheConnection> {
self.maybe_redis_ttl_cache
.as_ref()
.map(|redis_ttl_cache| redis_ttl_cache.get_connection()
.map_err(|err| {
warn!("redis cache failure: {:?}", err); // NB: We'll fail open if Redis cache fails
err
})
.ok())
.flatten()
}
}
@@ -6,7 +6,7 @@
use actix_web::HttpRequest;
use actix_web::cookie::Cookie;
use anyhow::anyhow;
use container_common::anyhow_result::AnyhowResult;
use errors::AnyhowResult;
use hmac::Hmac;
use hmac::NewMac;
use jwt::SignWithKey;
@@ -0,0 +1,39 @@
use actix_web::HttpRequest;
use billing_component::stripe::traits::internal_session_cache_purge::InternalSessionCachePurge;
use redis_caching::redis_ttl_cache::RedisTtlCache;
use redis_common::redis_cache_keys::RedisCacheKeys;
use users_component::utils::session_checker::SessionChecker;
pub struct InternalSessionCachePurgeImpl {
session_checker: SessionChecker,
redis_ttl_cache: RedisTtlCache,
}
impl InternalSessionCachePurgeImpl {
pub fn new(session_checker: SessionChecker, redis_ttl_cache: RedisTtlCache) -> Self {
Self {
session_checker,
redis_ttl_cache,
}
}
}
impl InternalSessionCachePurge for InternalSessionCachePurgeImpl {
fn best_effort_purge_session_cache(&self, http_request: &HttpRequest) {
// TODO: Clear Redis cache of sessions
// Unfortunately we don't yet have an index of user_token => session_tokens[] outside the DB.
// For now, a hacky solution is just to delete the cache under the current user.
// This makes sense for non-mods and should solve 95% of cases.
if let Some(session_token) = self.session_checker.forgiving_get_session_token(&http_request) {
if let Ok(mut redis_ttl_cache) = self.redis_ttl_cache.get_connection() {
let keys = vec![
RedisCacheKeys::session_record_user(&session_token),
RedisCacheKeys::session_record_light(&session_token),
];
for key in keys.iter() {
let _r = redis_ttl_cache.delete_from_cache(key).ok();
}
}
}
}
}
@@ -1,3 +1,4 @@
pub mod internal_product_to_stripe_lookup_impl;
pub mod internal_session_cache_purge_impl;
pub mod stripe_internal_subscription_product_lookup_impl;
pub mod stripe_internal_user_lookup_impl;
@@ -1,7 +1,7 @@
// NB: Incrementally getting rid of build warnings...
//#![forbid(unused_imports)]
//#![forbid(unused_mut)]
//#![forbid(unused_variables)]
#![forbid(unused_imports)]
#![forbid(unused_mut)]
#![forbid(unused_variables)]
use actix_web::error::ResponseError;
use actix_web::http::StatusCode;
@@ -10,14 +10,13 @@ use actix_web::{web, HttpResponse, HttpRequest};
use chrono::{DateTime, Utc};
use crate::server_state::ServerState;
use database_queries::column_types::vocoder_type::VocoderType;
use database_queries::queries::tts::tts_models::get_tts_model::{get_tts_model_by_token_using_connection, TtsModelRecord};
use database_queries::queries::tts::tts_models::get_tts_model::get_tts_model_by_token_using_connection;
use enums::common::visibility::Visibility;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
use log::warn;
use redis_common::redis_cache_keys::RedisCacheKeys;
use std::fmt;
use std::sync::Arc;
use errors::AnyhowResult;
use redis_common::redis_cache_keys::RedisCacheKeys;
use tts_common::text_pipelines::guess_pipeline::guess_text_pipeline_heuristic;
use tts_common::text_pipelines::text_pipeline_type::TextPipelineType;
+12 -1
View File
@@ -89,11 +89,13 @@ use actix_helpers::middleware::endpoint_disablement::disabled_endpoints::disable
use actix_helpers::middleware::endpoint_disablement::disabled_endpoints::exact_match_endpoint_disablements::ExactMatchEndpointDisablements;
use actix_helpers::middleware::endpoint_disablement::disabled_endpoints::prefix_endpoint_disablements::PrefixEndpointDisablements;
use actix_helpers::middleware::endpoint_disablement::endpoint_disablement_middleware::EndpointDisablementFilter;
use billing_component::stripe::traits::internal_session_cache_purge::InternalSessionCachePurge;
use redis_caching::redis_ttl_cache::RedisTtlCache;
use twitch_common::twitch_secrets::TwitchSecrets;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
use users_component::utils::session_checker::SessionChecker;
use users_component::utils::session_cookie_manager::SessionCookieManager;
use crate::billing::internal_session_cache_purge_impl::InternalSessionCachePurgeImpl;
const DEFAULT_BIND_ADDRESS : &'static str = "0.0.0.0:12345";
@@ -233,7 +235,10 @@ async fn main() -> AnyhowResult<()> {
easyenv::get_env_string_or_default("WEBSITE_HOMEPAGE_REDIRECT", "https://vo.codes/");
let cookie_manager = SessionCookieManager::new(&cookie_domain, &hmac_secret);
let session_checker = SessionChecker::new(&cookie_manager);
let session_checker = SessionChecker::new_with_cache(
&cookie_manager,
redis_ttl_cache.clone(),
);
let access_key = easyenv::get_env_string_required(ENV_ACCESS_KEY)?;
let secret_key = easyenv::get_env_string_required(ENV_SECRET_KEY)?;
@@ -508,12 +513,17 @@ pub async fn serve(server_state: ServerState) -> AnyhowResult<()>
server_state_arc.session_checker.clone(),
server_state_arc.mysql_pool.clone(),
));
let session_cache_purge : Arc<dyn InternalSessionCachePurge> = Arc::new(InternalSessionCachePurgeImpl::new(
server_state_arc.session_checker.clone(),
server_state_arc.redis_ttl_cache.clone(),
));
// NB: app_data being clone()'d below should all be safe (dependencies included)
let app = App::new()
.app_data(web::Data::new(server_state_arc.firehose_publisher.clone()))
.app_data(web::Data::new(server_state_arc.mysql_pool.clone()))
.app_data(web::Data::new(server_state_arc.redis_pool.clone()))
.app_data(web::Data::new(server_state_arc.redis_ttl_cache.clone()))
.app_data(web::Data::new(server_state_arc.session_checker.clone()))
.app_data(web::Data::new(server_state_arc.cookie_manager.clone()))
.app_data(web::Data::new(server_state_arc.stripe.clone().config.clone()))
@@ -523,6 +533,7 @@ pub async fn serve(server_state: ServerState) -> AnyhowResult<()>
.app_data(web::Data::from(product_lookup)) // NB: Data::from(Arc<T>) for dynamic dispatch
.app_data(web::Data::from(stripe_lookup)) // NB: Data::from(Arc<T>) for dynamic dispatch
.app_data(web::Data::from(user_lookup)) // NB: Data::from(Arc<T>) for dynamic dispatch
.app_data(web::Data::from(session_cache_purge)) // NB: Data::from(Arc<T>) for dynamic dispatch
.app_data(server_state_arc.clone())
.wrap(build_cors_config(server_state_arc.server_environment.clone()))
.wrap(DefaultHeaders::new()