diff --git a/Cargo.lock b/Cargo.lock index ac75cba236..50145adb42 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index c52346b3cc..41dc77b656 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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", diff --git a/crates/lib/database_queries/Cargo.toml b/crates/lib/database_queries/Cargo.toml index 3218d97956..31035bbf4b 100644 --- a/crates/lib/database_queries/Cargo.toml +++ b/crates/lib/database_queries/Cargo.toml @@ -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 diff --git a/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token.rs b/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token.rs index 248f230ef3..0fe46a9549 100644 --- a/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token.rs +++ b/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token.rs @@ -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, session_token: &str, ) -> AnyhowResult> { + 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> + 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 { diff --git a/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token_light.rs b/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token_light.rs index 92ddde72c3..62a4c69bdd 100644 --- a/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token_light.rs +++ b/crates/lib/database_queries/src/queries/users/user_sessions/get_user_session_by_token_light.rs @@ -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, - session_token: &str, -) -> AnyhowResult> { - // 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> + 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 { diff --git a/crates/lib/redis_common/src/redis_cache_keys.rs b/crates/lib/redis_common/src/redis_cache_keys.rs index a5f5a33523..5ec9e77a5f 100644 --- a/crates/lib/redis_common/src/redis_cache_keys.rs +++ b/crates/lib/redis_common/src/redis_cache_keys.rs @@ -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) + } } diff --git a/crates/lib/sqlx_mysql_helpers/Cargo.toml b/crates/lib/sqlx_mysql_helpers/Cargo.toml new file mode 100644 index 0000000000..25bd486fd3 --- /dev/null +++ b/crates/lib/sqlx_mysql_helpers/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "sqlx_mysql_helpers" +edition = "2018" +version = "0.0.1" +authors = [ + "Brandon Thomas ", + "Brandon Thomas ", +] +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 diff --git a/crates/lib/sqlx_mysql_helpers/src/any_connection.rs b/crates/lib/sqlx_mysql_helpers/src/any_connection.rs new file mode 100644 index 0000000000..cea51ad650 --- /dev/null +++ b/crates/lib/sqlx_mysql_helpers/src/any_connection.rs @@ -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), +} + +/// 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), +} + +/// Pass a handle to MySql without caring about the type. +pub enum MySqlHandleRef<'a> { + ConnectionPool(&'a MySqlPool), + Connection(&'a mut PoolConnection), +} + +//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> { + 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>, +// reference: &'a mut PoolConnection, +//} + + + +// TODO: +//impl <'a> MySqlConnectionRef<'a> { +// pub async fn get_pool_connection(&'a mut self) -> AnyhowResult<&'a PoolConnection> { +// match self { +// MySqlConnectionRef::ConnectionPool(pool) => { +// let connection = pool.acquire().await?; +// Ok(connection) +// } +// MySqlConnectionRef::Connection(inner) => Ok(inner), +// } +// +// } +//} diff --git a/crates/lib/sqlx_mysql_helpers/src/lib.rs b/crates/lib/sqlx_mysql_helpers/src/lib.rs new file mode 100644 index 0000000000..db53bf0c13 --- /dev/null +++ b/crates/lib/sqlx_mysql_helpers/src/lib.rs @@ -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; \ No newline at end of file diff --git a/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_json_handler.rs b/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_json_handler.rs index 18dbfdb51e..da018e0ca6 100644 --- a/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_json_handler.rs +++ b/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_json_handler.rs @@ -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, internal_product_to_stripe_lookup: web::Data, internal_user_lookup: web::Data, + internal_session_cache_purge: web::Data, ) -> Result { 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, diff --git a/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_redirect_handler.rs b/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_redirect_handler.rs index 878c226f3b..460bb18319 100644 --- a/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_redirect_handler.rs +++ b/crates/service/plugins/billing/src/stripe/http_endpoints/checkout/create/stripe_create_checkout_session_redirect_handler.rs @@ -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, internal_product_to_stripe_lookup: web::Data, internal_user_lookup: web::Data, + internal_session_cache_purge: web::Data, ) -> Result { 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()) diff --git a/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_json_handler.rs b/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_json_handler.rs index dabb1be12b..2ed6dfe8e8 100644 --- a/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_json_handler.rs +++ b/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_json_handler.rs @@ -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, internal_product_to_stripe_lookup: web::Data, internal_user_lookup: web::Data, + internal_session_cache_purge: web::Data, ) -> Result { 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, diff --git a/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_redirect_handler.rs b/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_redirect_handler.rs index 66a856ecf3..e7141f1585 100644 --- a/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_redirect_handler.rs +++ b/crates/service/plugins/billing/src/stripe/http_endpoints/customer_portal/stripe_create_customer_portal_session_redirect_handler.rs @@ -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, internal_product_to_stripe_lookup: web::Data, internal_user_lookup: web::Data, + internal_session_cache_purge: web::Data, ) -> Result { 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, diff --git a/crates/service/plugins/billing/src/stripe/traits/internal_session_cache_purge.rs b/crates/service/plugins/billing/src/stripe/traits/internal_session_cache_purge.rs new file mode 100644 index 0000000000..d125c53d7f --- /dev/null +++ b/crates/service/plugins/billing/src/stripe/traits/internal_session_cache_purge.rs @@ -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); +} diff --git a/crates/service/plugins/billing/src/stripe/traits/mod.rs b/crates/service/plugins/billing/src/stripe/traits/mod.rs index 44d38b0eac..2cf3b026fa 100644 --- a/crates/service/plugins/billing/src/stripe/traits/mod.rs +++ b/crates/service/plugins/billing/src/stripe/traits/mod.rs @@ -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; \ No newline at end of file diff --git a/crates/service/plugins/users/Cargo.toml b/crates/service/plugins/users/Cargo.toml index 299063e3f7..0c7eb97c7d 100644 --- a/crates/service/plugins/users/Cargo.toml +++ b/crates/service/plugins/users/Cargo.toml @@ -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" } diff --git a/crates/service/plugins/users/src/endpoints/edit_profile_handler.rs b/crates/service/plugins/users/src/endpoints/edit_profile_handler.rs index a7312aeba5..7fb2994e35 100644 --- a/crates/service/plugins/users/src/endpoints/edit_profile_handler.rs +++ b/crates/service/plugins/users/src/endpoints/edit_profile_handler.rs @@ -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, request: web::Json, mysql_pool: web::Data, + redis_ttl_cache: web::Data, session_checker: web::Data, ) -> Result { @@ -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, }; diff --git a/crates/service/plugins/users/src/utils/session_checker.rs b/crates/service/plugins/users/src/utils/session_checker.rs index 1f5dc95717..e177f38b70 100644 --- a/crates/service/plugins/users/src/utils/session_checker.rs +++ b/crates/service/plugins/users/src/utils/session_checker.rs @@ -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, } 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 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> { + self.cookie_manager.decode_session_token_from_request(request) + } + + pub fn forgiving_get_session_token(&self, request: &HttpRequest) -> Option { + self.get_session_token(request).ok().flatten() + } + + // ==================== SessionRecord ==================== + + //#[deprecated = "Use the PoolConnection method instead of the MySqlPool one."] pub async fn maybe_get_session_light( &self, request: &HttpRequest, pool: &MySqlPool ) -> AnyhowResult> { - 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, ) -> AnyhowResult> + { + 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> + 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> + 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 method instead of the MySqlPool one."] pub async fn maybe_get_user_session( &self, @@ -58,24 +115,60 @@ impl SessionChecker { pool: &MySqlPool, ) -> AnyhowResult> { - 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, ) -> AnyhowResult> + { + 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> + 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> + 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 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 { + 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() + } } diff --git a/crates/service/plugins/users/src/utils/session_cookie_manager.rs b/crates/service/plugins/users/src/utils/session_cookie_manager.rs index 748be8faa6..e3e7b9abf3 100644 --- a/crates/service/plugins/users/src/utils/session_cookie_manager.rs +++ b/crates/service/plugins/users/src/utils/session_cookie_manager.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/billing/internal_session_cache_purge_impl.rs b/crates/service/web/storyteller_web/src/billing/internal_session_cache_purge_impl.rs new file mode 100644 index 0000000000..3563a22919 --- /dev/null +++ b/crates/service/web/storyteller_web/src/billing/internal_session_cache_purge_impl.rs @@ -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(); + } + } + } + } +} diff --git a/crates/service/web/storyteller_web/src/billing/mod.rs b/crates/service/web/storyteller_web/src/billing/mod.rs index c037329014..bf3689f826 100644 --- a/crates/service/web/storyteller_web/src/billing/mod.rs +++ b/crates/service/web/storyteller_web/src/billing/mod.rs @@ -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; \ No newline at end of file diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_tts_model.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_tts_model.rs index b60a01618b..916922bcb7 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_tts_model.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/tts/get_tts_model.rs @@ -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; diff --git a/crates/service/web/storyteller_web/src/main.rs b/crates/service/web/storyteller_web/src/main.rs index b599d4dd5e..401019d5a5 100644 --- a/crates/service/web/storyteller_web/src/main.rs +++ b/crates/service/web/storyteller_web/src/main.rs @@ -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 = 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) for dynamic dispatch .app_data(web::Data::from(stripe_lookup)) // NB: Data::from(Arc) for dynamic dispatch .app_data(web::Data::from(user_lookup)) // NB: Data::from(Arc) for dynamic dispatch + .app_data(web::Data::from(session_cache_purge)) // NB: Data::from(Arc) for dynamic dispatch .app_data(server_state_arc.clone()) .wrap(build_cors_config(server_state_arc.server_environment.clone())) .wrap(DefaultHeaders::new()