diff --git a/crates/lib/mysql_queries/src/queries/entity_stats/mod.rs b/crates/lib/mysql_queries/src/queries/entity_stats/mod.rs index 5e7fd9833d..8cf00322a3 100644 --- a/crates/lib/mysql_queries/src/queries/entity_stats/mod.rs +++ b/crates/lib/mysql_queries/src/queries/entity_stats/mod.rs @@ -1,2 +1,3 @@ pub mod stats_entity_token; -pub mod upsert_entity_stats_on_bookmark_event; \ No newline at end of file +pub mod upsert_entity_stats_on_bookmark_event; +pub mod upsert_entity_stats_on_ratings_event; \ No newline at end of file diff --git a/crates/lib/mysql_queries/src/queries/entity_stats/stats_entity_token.rs b/crates/lib/mysql_queries/src/queries/entity_stats/stats_entity_token.rs index 873efd8615..27347edbed 100644 --- a/crates/lib/mysql_queries/src/queries/entity_stats/stats_entity_token.rs +++ b/crates/lib/mysql_queries/src/queries/entity_stats/stats_entity_token.rs @@ -1,5 +1,6 @@ use enums::by_table::entity_stats::stats_entity_type::StatsEntityType; use enums::by_table::user_bookmarks::user_bookmark_entity_type::UserBookmarkEntityType; +use enums::by_table::user_ratings::entity_type::UserRatingEntityType; use tokens::tokens::comments::CommentToken; use tokens::tokens::media_files::MediaFileToken; use tokens::tokens::model_weights::ModelWeightToken; @@ -21,6 +22,15 @@ impl StatsEntityToken { } } + /// Rating entity types aren't 1:1, and they're not a superset (yet) either. + pub fn from_rating_entity_type_and_token(entity_type: UserRatingEntityType, token: &str) -> Option { + match entity_type { + UserRatingEntityType::MediaFile => Some(Self::MediaFile(MediaFileToken::new_from_str(token))), + UserRatingEntityType::ModelWeight => Some(Self::ModelWeight(ModelWeightToken::new_from_str(token))), + _ => None, + } + } + pub fn from_entity_type_and_token(entity_type: StatsEntityType, token: &str) -> Self { match entity_type { StatsEntityType::Comment => Self::Comment(CommentToken::new_from_str(token)), diff --git a/crates/lib/mysql_queries/src/queries/entity_stats/upsert_entity_stats_on_ratings_event.rs b/crates/lib/mysql_queries/src/queries/entity_stats/upsert_entity_stats_on_ratings_event.rs new file mode 100644 index 0000000000..92d81a2a0d --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/entity_stats/upsert_entity_stats_on_ratings_event.rs @@ -0,0 +1,152 @@ +use std::marker::PhantomData; + +use anyhow::anyhow; +use sqlx::{Executor, MySql}; + +use errors::AnyhowResult; + +use crate::queries::entity_stats::stats_entity_token::StatsEntityToken; + +/// NB: "Neutral" means one of two things here: +/// 1. The user has not voted on the entity, ie. no previous vote exists (null). +/// 2. The user has voted on the entity, but the vote is "neutral". +#[derive(Copy, Clone)] +pub enum RatingsAction { + /// Add one to positive + NeutralToPositive, + + /// Add one to negative + NeutralToNegative, + + /// Remove one from positive + PositiveToNeutral, + + /// Remove one from negative + NegativeToNeutral, + + /// Remove one from negative, add one to positive + NegativeToPositive, + + /// Remove one from positive, add one to negative + PositiveToNegative, +} + +pub struct UpsertEntityStatsArgs<'e, 'c, E> + where E: 'e + Executor<'c, Database = MySql> +{ + pub stats_entity_token: &'e StatsEntityToken, + pub action: RatingsAction, + pub mysql_executor: E, + + // TODO: Not sure if this works to tell the compiler we need the lifetime annotation. + // See: https://doc.rust-lang.org/std/marker/struct.PhantomData.html#unused-lifetime-parameters + pub phantom: PhantomData<&'c E>, +} + +pub async fn upsert_entity_stats_on_ratings_event<'e, 'c : 'e, E>( + args: UpsertEntityStatsArgs<'e, 'c, E>, +) -> AnyhowResult<()> + where E: 'e + Executor<'c, Database = MySql> +{ + let (entity_type, entity_token) = args.stats_entity_token.get_composite_keys(); + + let query = match args.action { + RatingsAction::NeutralToPositive => { + sqlx::query!( + r#" + INSERT INTO entity_stats SET + entity_type = ?, + entity_token = ?, + ratings_positive_count = 1 + ON DUPLICATE KEY UPDATE + ratings_positive_count = ratings_positive_count + 1 + "#, + entity_type, + entity_token, + ) + } + RatingsAction::NeutralToNegative => { + sqlx::query!( + r#" + INSERT INTO entity_stats SET + entity_type = ?, + entity_token = ?, + ratings_negative_count = 1 + ON DUPLICATE KEY UPDATE + ratings_negative_count = ratings_negative_count + 1 + "#, + entity_type, + entity_token, + ) + } + RatingsAction::PositiveToNeutral => { + sqlx::query!( + r#" + INSERT INTO entity_stats SET + entity_type = ?, + entity_token = ?, + ratings_positive_count = 0 + ON DUPLICATE KEY UPDATE + ratings_positive_count = ABS(CAST(ratings_positive_count AS SIGNED) - 1) + "#, + entity_type, + entity_token, + ) + } + RatingsAction::NegativeToNeutral => { + sqlx::query!( + r#" + INSERT INTO entity_stats SET + entity_type = ?, + entity_token = ?, + ratings_negative_count = 0 + ON DUPLICATE KEY UPDATE + ratings_negative_count = ABS(CAST(ratings_negative_count AS SIGNED) - 1) + "#, + entity_type, + entity_token, + ) + } + RatingsAction::PositiveToNegative => { + sqlx::query!( + r#" + INSERT INTO entity_stats SET + entity_type = ?, + entity_token = ?, + ratings_positive_count = 0, + ratings_negative_count = 1 + ON DUPLICATE KEY UPDATE + ratings_positive_count = ABS(CAST(ratings_positive_count AS SIGNED) - 1), + ratings_negative_count = ratings_negative_count + 1 + "#, + entity_type, + entity_token, + ) + } + RatingsAction::NegativeToPositive => { + sqlx::query!( + r#" + INSERT INTO entity_stats SET + entity_type = ?, + entity_token = ?, + ratings_positive_count = 1, + ratings_negative_count = 0 + ON DUPLICATE KEY UPDATE + ratings_positive_count = ratings_positive_count + 1, + ratings_negative_count = ABS(CAST(ratings_negative_count AS SIGNED) - 1) + "#, + entity_type, + entity_token, + ) + } + }; + + let query_result = query.execute(args.mysql_executor).await; + + let _record_id = match query_result { + Ok(res) => res.last_insert_id(), + Err(err) => return Err(anyhow!("Mysql error: {:?}", err)), + }; + + Ok(()) +} diff --git a/crates/lib/mysql_queries/src/queries/users/user_ratings/get_user_rating_transactional_locking.rs b/crates/lib/mysql_queries/src/queries/users/user_ratings/get_user_rating_transactional_locking.rs new file mode 100644 index 0000000000..e93cf8c68d --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/users/user_ratings/get_user_rating_transactional_locking.rs @@ -0,0 +1,85 @@ +use anyhow::anyhow; +use chrono::{DateTime, Utc}; +use log::error; +use sqlx::{Executor, MySql}; + +use enums::by_table::user_ratings::rating_value::UserRatingValue; +use errors::AnyhowResult; +use tokens::tokens::users::UserToken; + +use crate::composite_keys::by_table::user_ratings::user_rating_entity::UserRatingEntity; + +pub struct UserRating { + pub rating_value: UserRatingValue, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +/// Look up the user rating +pub async fn get_user_rating_transactional_locking<'e, 'c, E>( + user_token : &'e UserToken, + user_rating_entity: &'e UserRatingEntity, + mysql_executor: E, +) -> AnyhowResult> + where E: 'e + Executor<'c, Database = MySql> +{ + let entity_type = user_rating_entity.get_entity_type(); + let entity_token = user_rating_entity.get_entity_token_str(); + + // NB: LEFT OUTER JOIN does not require entity_stats to be present, but will lock it under + // SELECT...FOR UPDATE if the row exists. + let maybe_result = sqlx::query_as!( + InternalUserRatingRecord, + r#" +SELECT + r.rating_value as `rating_value: enums::by_table::user_ratings::rating_value::UserRatingValue`, + r.created_at, + r.updated_at, + + es.ratings_positive_count as unused_ratings_positive_count + +FROM user_ratings AS r + +LEFT OUTER JOIN entity_stats AS es + ON r.entity_type = es.entity_type + AND r.entity_token = es.entity_token + +WHERE r.user_token = ? + AND r.entity_type = ? + AND r.entity_token = ? + +LIMIT 1 + +FOR UPDATE + "#, + user_token.as_str(), + entity_type, + entity_token, + ) + .fetch_one(mysql_executor) + .await; + + match maybe_result { + Ok(record) => Ok(Some(UserRating { + rating_value: record.rating_value, + created_at: record.created_at, + updated_at: record.updated_at, + })), + Err(err) => match err { + sqlx::Error::RowNotFound => Ok(None), + _ => { + error!("error querying job record: {:?}", err); + Err(anyhow!("error querying job record: {:?}", err)) + } + } + } +} + +struct InternalUserRatingRecord { + rating_value: UserRatingValue, + created_at: DateTime, + updated_at: DateTime, + + // NB: Not used, just for dual transactional locking. + pub unused_ratings_positive_count: Option, +} diff --git a/crates/lib/mysql_queries/src/queries/users/user_ratings/mod.rs b/crates/lib/mysql_queries/src/queries/users/user_ratings/mod.rs index aabf007efb..2be63ff694 100644 --- a/crates/lib/mysql_queries/src/queries/users/user_ratings/mod.rs +++ b/crates/lib/mysql_queries/src/queries/users/user_ratings/mod.rs @@ -1,5 +1,6 @@ pub mod batch_get_user_rating; pub mod get_total_user_rating_count_for_entity; pub mod get_user_rating; +pub mod get_user_rating_transactional_locking; pub mod update_tts_model_ratings; pub mod upsert_user_rating; \ No newline at end of file diff --git a/crates/lib/mysql_queries/src/queries/users/user_ratings/upsert_user_rating.rs b/crates/lib/mysql_queries/src/queries/users/user_ratings/upsert_user_rating.rs index 4be3f29197..0c0c6119e7 100644 --- a/crates/lib/mysql_queries/src/queries/users/user_ratings/upsert_user_rating.rs +++ b/crates/lib/mysql_queries/src/queries/users/user_ratings/upsert_user_rating.rs @@ -1,5 +1,6 @@ -use sqlx::MySql; -use sqlx::pool::PoolConnection; +use std::marker::PhantomData; + +use sqlx::{Executor, MySql}; use enums::by_table::user_ratings::rating_value::UserRatingValue; use errors::AnyhowResult; @@ -7,15 +8,25 @@ use tokens::tokens::users::UserToken; use crate::composite_keys::by_table::user_ratings::user_rating_entity::UserRatingEntity; -pub struct Args<'a> { - pub user_token: &'a UserToken, - pub user_rating_entity: &'a UserRatingEntity, +pub struct Args<'e, 'c, E> + where E: 'e + Executor<'c, Database = MySql> +{ + pub user_token: &'e UserToken, + pub user_rating_entity: &'e UserRatingEntity, pub user_rating_value: UserRatingValue, - pub ip_address: &'a str, - pub mysql_connection: &'a mut PoolConnection, + pub ip_address: &'e str, + pub mysql_executor: E, + + // TODO: Not sure if this works to tell the compiler we need the lifetime annotation. + // See: https://doc.rust-lang.org/std/marker/struct.PhantomData.html#unused-lifetime-parameters + pub phantom: PhantomData<&'c E>, } -pub async fn upsert_user_rating(args: Args<'_>) -> AnyhowResult<()> { +pub async fn upsert_user_rating<'e, 'c : 'e, E>( + args: Args<'e, 'c, E>, +) -> AnyhowResult<()> + where E: 'e + Executor<'c, Database = MySql> +{ let entity_type = args.user_rating_entity.get_entity_type(); let entity_token = args.user_rating_entity.get_entity_token_str(); @@ -45,6 +56,7 @@ ON DUPLICATE KEY UPDATE args.ip_address ); - let _r = query.execute(&mut **args.mysql_connection).await?; + let _r = query.execute(args.mysql_executor).await?; + Ok(()) } diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/user_bookmarks/create_user_bookmark_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/user_bookmarks/create_user_bookmark_handler.rs index d3d0443670..638ec1500b 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/user_bookmarks/create_user_bookmark_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/user_bookmarks/create_user_bookmark_handler.rs @@ -210,6 +210,7 @@ pub async fn create_user_bookmark_handler( })?; // TODO(bt,2024-01-04): The methods of stats collection here differs. + // Update this to return directly from the stats table instead of doing a COUNT(*). let count = get_total_bookmark_count_for_entity(&entity_token, &mut mysql_connection) .await diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/user_ratings/set_user_rating_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/user_ratings/set_user_rating_handler.rs index 30fe8afd02..26e3fd6467 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/user_ratings/set_user_rating_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/user_ratings/set_user_rating_handler.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use actix_web::{HttpRequest, HttpResponse, ResponseError, web}; use actix_web::http::StatusCode; use log::{error, info, warn}; +use sqlx::Acquire; use utoipa::ToSchema; use enums::by_table::user_ratings::entity_type::UserRatingEntityType; @@ -10,7 +11,10 @@ use enums::by_table::user_ratings::rating_value::UserRatingValue; 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 mysql_queries::composite_keys::by_table::user_ratings::user_rating_entity::UserRatingEntity; +use mysql_queries::queries::entity_stats::stats_entity_token::StatsEntityToken; +use mysql_queries::queries::entity_stats::upsert_entity_stats_on_ratings_event::{RatingsAction, upsert_entity_stats_on_ratings_event, UpsertEntityStatsArgs}; use mysql_queries::queries::users::user_ratings::get_total_user_rating_count_for_entity::get_total_user_rating_count_for_entity; +use mysql_queries::queries::users::user_ratings::get_user_rating_transactional_locking::get_user_rating_transactional_locking; use mysql_queries::queries::users::user_ratings::update_tts_model_ratings::update_tts_model_ratings; use mysql_queries::queries::users::user_ratings::upsert_user_rating::{Args, upsert_user_rating}; use tokens::tokens::media_files::MediaFileToken; @@ -156,12 +160,29 @@ pub async fn set_user_rating_handler( return Err(SetUserRatingError::BadInput("type not yet supported".to_string())), }; + let mut transaction = mysql_connection.begin().await + .map_err(|err| { + error!("error creating transaction: {:?}", err); + SetUserRatingError::ServerError + })?; + + let maybe_existing_user_rating = get_user_rating_transactional_locking( + &user_session.user_token_typed, + &entity, + &mut *transaction, + ).await + .map_err(|err| { + error!("error getting user rating: {:?}", err); + SetUserRatingError::ServerError + })?; + let _r = upsert_user_rating(Args { user_token: &user_session.user_token_typed, user_rating_entity: &entity, user_rating_value: request.rating_value, ip_address: &ip_address, - mysql_connection: &mut mysql_connection + mysql_executor: &mut *transaction, + phantom: Default::default(), }) .await .map_err(|err| { @@ -169,6 +190,51 @@ pub async fn set_user_rating_handler( SetUserRatingError::ServerError })?; + let existing_rating_value = maybe_existing_user_rating + .map(|rating| rating.rating_value) + .unwrap_or(UserRatingValue::Neutral); + + let mut maybe_rating_action = + match (existing_rating_value, request.rating_value) { + (UserRatingValue::Neutral, UserRatingValue::Neutral) => None, + (UserRatingValue::Neutral, UserRatingValue::Positive) => Some(RatingsAction::NeutralToPositive), + (UserRatingValue::Neutral, UserRatingValue::Negative) => Some(RatingsAction::NeutralToNegative), + (UserRatingValue::Positive, UserRatingValue::Neutral) => Some(RatingsAction::PositiveToNeutral), + (UserRatingValue::Positive, UserRatingValue::Positive) => None, + (UserRatingValue::Positive, UserRatingValue::Negative) => Some(RatingsAction::PositiveToNegative), + (UserRatingValue::Negative, UserRatingValue::Neutral) => Some(RatingsAction::NegativeToNeutral), + (UserRatingValue::Negative, UserRatingValue::Positive) => Some(RatingsAction::NeutralToPositive), + (UserRatingValue::Negative, UserRatingValue::Negative) => None, + }; + + if let Some(rating_action) = maybe_rating_action { + + // NB: Not all rateable things have stats (eg. deprecated record types don't have stats). + let maybe_stats_entity_token = + StatsEntityToken::from_rating_entity_type_and_token(request.entity_type, &request.entity_token); + + if let Some(stats_entity_token) = maybe_stats_entity_token { + upsert_entity_stats_on_ratings_event(UpsertEntityStatsArgs { + stats_entity_token: &stats_entity_token, + action: rating_action, + mysql_executor: &mut *transaction, + phantom: Default::default(), + }) + .await + .map_err(|err| { + error!("Error upserting entity stats: {:?}", err); + SetUserRatingError::ServerError + })?; + } + } + + transaction.commit().await + .map_err(|err| { + error!("error committing transaction: {:?}", err); + SetUserRatingError::ServerError + })?; + + // NB: Legacy match request.entity_type { UserRatingEntityType::TtsModel => { let token = TtsModelToken::new_from_str(&request.entity_token); @@ -184,6 +250,9 @@ pub async fn set_user_rating_handler( } } + // TODO(bt,2024-01-04): The methods of stats collection here differs. + // Update this to return directly from the stats table instead of doing a COUNT(*). + let count = get_total_user_rating_count_for_entity(&entity, &mut mysql_connection) .await .map_err(|err| {