collect rating stats

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;
+pub mod upsert_entity_stats_on_bookmark_event;
+pub mod upsert_entity_stats_on_ratings_event;

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<Self> {
+    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)),

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(())
+}

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<Utc>,
+  pub updated_at: DateTime<Utc>,
+}
+
+/// 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<Option<UserRating>>
+  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<Utc>,
+  updated_at: DateTime<Utc>,
+
+  // NB: Not used, just for dual transactional locking.
+  pub unused_ratings_positive_count: Option<u32>,
+}

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;

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<MySql>,
+  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(())
 }

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

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| {