Merge pull request #27 from storytold/bt-trending-voices

Bt trending voices

Cargo.lock

@@ -640,6 +640,33 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "analytics-job"
+version = "0.0.1"
+dependencies = [
+ "anyhow",
+ "chrono",
+ "clap 2.34.0",
+ "config",
+ "container_common",
+ "database_queries",
+ "dotenv",
+ "easyenv",
+ "enums",
+ "errors",
+ "hostname",
+ "jobs_common",
+ "log",
+ "rand 0.8.4",
+ "reusable_types",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "sqlx",
+ "tokens",
+ "tokio 1.16.1",
+]
+
 [[package]]
 name = "ansi_term"
 version = "0.12.1"

Cargo.toml

@@ -39,6 +39,7 @@ members = [
   "crates/lib/user_input_common",
 
   # Services: API servers, jobs, workers, etc.
+  "crates/service/job/analytics_job",
   "crates/service/job/download_job",
   "crates/service/job/inference_job",
   "crates/service/job/tts_download_job",

Dockerfile

@@ -78,6 +78,12 @@ RUN SQLX_OFFLINE=true \
   --release \
   --bin storyteller-web
 
+RUN SQLX_OFFLINE=true \
+  LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \
+  $HOME/.cargo/bin/cargo build \
+  --release \
+  --bin analytics-job \
+
 RUN SQLX_OFFLINE=true \
   LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \
   $HOME/.cargo/bin/cargo build \
@@ -149,6 +155,7 @@ RUN echo -n ${GIT_SHA} > GIT_SHA
 
 # Copy all the binaries.
 COPY --from=builder /tmp/target/release/storyteller-web /
+COPY --from=builder /tmp/target/release/analytics-job /
 COPY --from=builder /tmp/target/release/download-job /
 COPY --from=builder /tmp/target/release/inference-job /
 COPY --from=builder /tmp/target/release/tts-download-job /

_migrations/2023-01-16-211642_trending_model_analytics/down.sql

@@ -0,0 +1,5 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+DROP TABLE trending_model_analytics;

_migrations/2023-01-16-211642_trending_model_analytics/up.sql

@@ -0,0 +1,40 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+-- Periodically capture analytics on trending models
+CREATE TABLE trending_model_analytics (
+  -- Not used for anything except replication.
+  id BIGINT(20) NOT NULL AUTO_INCREMENT,
+
+  -- The type of model. TTS, VC, etc.
+  model_type VARCHAR(32) NOT NULL,
+
+  -- The primary key of the TTS, VC, or other model in question.
+  model_token VARCHAR(32) NOT NULL,
+
+  -- The name of the window, eg. "last_24_hours", "last_5_hours", etc.
+  window_name VARCHAR(32) NOT NULL,
+
+  -- The measured numerical value of the statistic, eg "500 uses" = 500.
+  numeric_value INT(10) NOT NULL DEFAULT 0,
+
+  -- ========== VECTOR CLOCK ==========
+
+  -- Incremented with every update.
+  version INT NOT NULL DEFAULT 0,
+
+  -- ========== TIMESTAMPS ==========
+
+  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+  -- INDICES --
+  PRIMARY KEY (id),
+  UNIQUE KEY (model_token, model_type, window_name),
+  KEY index_model_type_model_token (model_type, model_token),
+  KEY index_model_type (model_type),
+  KEY index_model_token (model_token),
+  KEY index_numeric_value (numeric_value)
+
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

analytics/tts_models.sql

@@ -7,8 +7,9 @@
 --
 select
     m.token,
-    m.title,
+    m.text_pipeline_type,
     m.ietf_language_tag,
+    m.title,
     u.username,
     r.use_count,
     m.user_deleted_at,
@@ -37,25 +38,18 @@ select m.token, m.title, r.use_count from (
   join tts_models as m
   on m.token = r.model_token;
 
---
--- Top 100 models by use count, last 30 days
---
-select m.token, m.title, m.ietf_language_tag, r.use_count from (
-    select model_token, count(*) as use_count
-    from tts_results
-    where created_at > ( CURDATE() - INTERVAL 30 DAY )
-    group by model_token
-    order by use_count desc
-        limit 500
-) as r
-    join tts_models as m
-    on m.token = r.model_token;
-
 --
 -- Top 500 models by use count, last 5 days
 -- Limited due to tmux scrollback.
 --
-select m.token, m.title, m.ietf_language_tag, u.username, r.use_count from (
+select
+    m.token,
+    m.text_pipeline_type,
+    m.ietf_language_tag,
+    m.title,
+    u.username,
+    r.use_count
+from (
     select model_token, count(*) as use_count
     from tts_results
     where created_at > ( CURDATE() - INTERVAL 5 DAY )
@@ -74,8 +68,9 @@ order by r.use_count desc
 --
 select
     m.token,
-    m.title,
+    m.text_pipeline_type,
     m.ietf_language_tag,
+    m.title,
     u.username,
     r.use_count,
     m.user_deleted_at,

crates/_types/enums/src/by_table/mod.rs

@@ -1,2 +1,3 @@
 pub mod media_uploads;
 pub mod model_categories;
+pub mod trending_model_analytics;

crates/_types/enums/src/by_table/trending_model_analytics/mod.rs

@@ -0,0 +1,2 @@
+pub mod model_type;
+pub mod window_name;

crates/_types/enums/src/by_table/trending_model_analytics/model_type.rs

@@ -0,0 +1,63 @@
+use serde::Deserialize;
+use serde::Serialize;
+
+#[cfg(test)]
+use strum::EnumCount;
+#[cfg(test)]
+use strum::EnumIter;
+
+/// Used in the `trending_model_analytics` table in a `VARCHAR` field.
+///
+/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY.
+#[cfg_attr(test, derive(EnumIter, EnumCount))]
+#[derive(Clone, Copy, Eq, PartialEq, Deserialize, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ModelType {
+  /// TTS models
+  Tts,
+
+  // NB: It's assumed we'll use this same system to track
+  // trending VC models too, so the next type would be "VC".
+}
+
+// TODO(bt, 2023-01-17): This desperately needs MySQL integration tests!
+impl_enum_display_and_debug_using_to_str!(ModelType);
+impl_enum_sqlx_coders!(ModelType);
+
+/// NB: Legacy API for older code.
+impl ModelType {
+  pub fn to_str(&self) -> &'static str {
+    match self {
+      Self::Tts => "tts",
+    }
+  }
+
+  pub fn from_str(value: &str) -> Result<Self, String> {
+    match value {
+      "tts" => Ok(Self::Tts),
+      _ => Err(format!("invalid value: {:?}", value)),
+    }
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use crate::by_table::trending_model_analytics::model_type::ModelType;
+  use crate::test_helpers::assert_serialization;
+
+  #[test]
+  fn test_serialization() {
+    assert_serialization(ModelType::Tts, "tts");
+  }
+
+  #[test]
+  fn test_to_str() {
+    assert_eq!(ModelType::Tts.to_str(), "tts");
+  }
+
+  #[test]
+  fn test_from_str() {
+    assert_eq!(ModelType::from_str("tts").unwrap(), ModelType::Tts);
+    assert!(ModelType::from_str("foo").is_err());
+  }
+}

crates/_types/enums/src/by_table/trending_model_analytics/window_name.rs

@@ -0,0 +1,88 @@
+use serde::Deserialize;
+use serde::Serialize;
+
+#[cfg(test)]
+use strum::EnumCount;
+#[cfg(test)]
+use strum::EnumIter;
+
+/// Used in the `window_name` table in a `VARCHAR` field.
+///
+/// Contrary to most of this crate and unlike most "enum"-types
+/// that are inflexible, new window names can be added/removed
+/// without breaking too much.
+#[cfg_attr(test, derive(EnumIter, EnumCount))]
+#[derive(Clone, Copy, Eq, PartialEq, Hash, Deserialize, Serialize)]
+pub enum WindowName {
+  /// Analytics over the last three hours
+  #[serde(rename = "last_3_hours")]
+  Last3Hours,
+
+  /// Analytics over the last three hours
+  #[serde(rename = "last_3_days")]
+  Last3Days,
+
+  /// Analytics over all historical records
+  #[serde(rename = "all_time")]
+  AllTime,
+}
+
+// TODO(bt, 2023-01-17): This desperately needs MySQL integration tests!
+impl_enum_display_and_debug_using_to_str!(WindowName);
+impl_enum_sqlx_coders!(WindowName);
+
+/// NB: Legacy API for older code.
+impl WindowName {
+  pub fn to_str(&self) -> &'static str {
+    match self {
+      Self::Last3Hours => "last_3_hours",
+      Self::Last3Days => "last_3_days",
+      Self::AllTime => "all_time",
+    }
+  }
+
+  pub fn from_str(value: &str) -> Result<Self, String> {
+    match value {
+      "last_3_hours" => Ok(Self::Last3Hours),
+      "last_3_days" => Ok(Self::Last3Days),
+      "all_time" => Ok(Self::AllTime),
+      _ => Err(format!("invalid value: {:?}", value)),
+    }
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use crate::by_table::trending_model_analytics::window_name::WindowName;
+  use crate::test_helpers::assert_serialization;
+
+  mod serde {
+    use super::*;
+
+    #[test]
+    fn test_serialization() {
+      assert_serialization(WindowName::Last3Hours, "last_3_hours");
+      assert_serialization(WindowName::Last3Days, "last_3_days");
+      assert_serialization(WindowName::AllTime, "all_time");
+    }
+  }
+
+  mod impl_methods {
+    use super::*;
+
+    #[test]
+    fn test_to_str() {
+      assert_eq!(WindowName::Last3Hours.to_str(), "last_3_hours");
+      assert_eq!(WindowName::Last3Days.to_str(), "last_3_days");
+      assert_eq!(WindowName::AllTime.to_str(), "all_time");
+    }
+
+    #[test]
+    fn test_from_str() {
+      assert_eq!(WindowName::from_str("last_3_hours").unwrap(), WindowName::Last3Hours);
+      assert_eq!(WindowName::from_str("last_3_days").unwrap(), WindowName::Last3Days);
+      assert_eq!(WindowName::from_str("all_time").unwrap(), WindowName::AllTime);
+      assert!(WindowName::from_str("foo").is_err());
+    }
+  }
+}

crates/lib/database_queries/src/queries/mod.rs

@@ -7,6 +7,7 @@ pub mod ip_bans;
 pub mod media_uploads;
 pub mod model_categories;
 pub mod stats;
+pub mod trending_model_analytics;
 pub mod tts;
 pub mod twitch;
 pub mod users;

crates/lib/database_queries/src/queries/trending_model_analytics/list_trending_tts_models.rs

@@ -0,0 +1,117 @@
+use sqlx::{MySql, MySqlPool};
+use sqlx::pool::PoolConnection;
+use enums::by_table::trending_model_analytics::model_type::ModelType;
+use enums::by_table::trending_model_analytics::window_name::WindowName;
+use errors::AnyhowResult;
+use tokens::tokens::tts_models::TtsModelToken;
+
+pub struct TrendingModels {
+  pub models: Vec<TrendingModel>,
+}
+
+pub struct TrendingModel {
+  pub model_token: TtsModelToken,
+
+  // NB: "en", "es", etc. not "en-US", "es-419", etc.
+  pub ietf_primary_language_subtag: String,
+
+  // Time window
+  pub window_name: WindowName,
+
+  // Count of uses within the window.
+  pub numeric_value: u64,
+}
+
+// TODO(bt,2023-01-17): This query is massive, so is not type checked :(
+
+pub async fn list_trending_tts_models_with_pool(
+  mysql_pool: &MySqlPool,
+) -> AnyhowResult<TrendingModels> {
+  let mut mysql_connection = mysql_pool.acquire().await?;
+  list_trending_tts_models(&mut mysql_connection).await
+}
+
+pub async fn list_trending_tts_models(
+  mysql_connection: &mut PoolConnection<MySql>,
+) -> AnyhowResult<TrendingModels> {
+  let query_parts = vec![
+    // English
+    make_subquery(ModelType::Tts, WindowName::AllTime, "en"),
+    make_subquery(ModelType::Tts, WindowName::Last3Hours, "en"),
+    make_subquery(ModelType::Tts, WindowName::Last3Days, "en"),
+
+    // Spanish
+    make_subquery(ModelType::Tts, WindowName::AllTime, "es"),
+    make_subquery(ModelType::Tts, WindowName::Last3Hours, "es"),
+    make_subquery(ModelType::Tts, WindowName::Last3Days, "es"),
+
+    // Italian
+    make_subquery(ModelType::Tts, WindowName::AllTime, "it"),
+    make_subquery(ModelType::Tts, WindowName::Last3Hours, "it"),
+    make_subquery(ModelType::Tts, WindowName::Last3Days, "it"),
+
+    // French
+    make_subquery(ModelType::Tts, WindowName::AllTime, "fr"),
+    make_subquery(ModelType::Tts, WindowName::Last3Hours, "fr"),
+    make_subquery(ModelType::Tts, WindowName::Last3Days, "fr"),
+
+    // German
+    make_subquery(ModelType::Tts, WindowName::AllTime, "de"),
+    make_subquery(ModelType::Tts, WindowName::Last3Hours, "de"),
+    make_subquery(ModelType::Tts, WindowName::Last3Days, "de"),
+  ];
+
+  let query = query_parts.join(" UNION ");
+  let mut query = sqlx::query_as::<_, RawTrendingModel>(&query);
+
+  let results = query.fetch_all(mysql_connection).await?;
+
+  let results = results.into_iter()
+      .map(|model| TrendingModel {
+        model_token: TtsModelToken::new(model.model_token),
+        ietf_primary_language_subtag: model.ietf_primary_language_subtag,
+        window_name: WindowName::from_str(&model.window_name).unwrap_or(WindowName::AllTime), // TODO: Fail properly
+        numeric_value: model.numeric_value as u64,
+      })
+      .collect::<Vec<TrendingModel>>();
+
+  Ok(TrendingModels {
+    models: results,
+  })
+}
+
+// Build queries meant to be UNION-d together.
+pub fn make_subquery(model_type: ModelType, window_name: WindowName, ietf_primary_language_subtag: &str) -> String {
+  format!(r#"
+(
+  SELECT
+    m.token as model_token,
+    m.ietf_primary_language_subtag,
+    t.window_name,
+    t.numeric_value
+
+  FROM trending_model_analytics AS t
+  JOIN tts_models AS m
+  ON m.token = t.model_token
+
+  WHERE
+    t.model_type = "{}"
+    AND t.window_name = "{}"
+    AND m.ietf_primary_language_subtag = "{}"
+
+  ORDER BY numeric_value DESC
+  LIMIT 10
+)
+  "#,
+          model_type,
+          window_name,
+          ietf_primary_language_subtag)
+}
+
+#[derive(sqlx::FromRow)]
+struct RawTrendingModel {
+  pub model_token: String,
+  pub ietf_primary_language_subtag: String,
+  pub window_name: String,
+  pub numeric_value: i64,
+}

crates/lib/database_queries/src/queries/trending_model_analytics/mod.rs

@@ -0,0 +1,2 @@
+pub mod list_trending_tts_models;
+pub mod upsert_trending_model_analytics;

crates/lib/database_queries/src/queries/trending_model_analytics/upsert_trending_model_analytics.rs

@@ -0,0 +1,50 @@
+use container_common::anyhow_result::AnyhowResult;
+use enums::by_table::trending_model_analytics::model_type::ModelType;
+use enums::by_table::trending_model_analytics::window_name::WindowName;
+use sqlx::MySql;
+use sqlx::pool::PoolConnection;
+use tokens::tokens::tts_models::TtsModelToken;
+
+// NB: Only TTS for now. New enum variants for other token types.
+pub enum ModelToken<'a> {
+  Tts(&'a TtsModelToken),
+}
+
+pub struct Args<'a> {
+  pub model_token: ModelToken<'a>,
+  pub window_name: WindowName,
+  pub numeric_value: u64,
+  pub mysql_connection: &'a mut PoolConnection<MySql>,
+}
+
+pub async fn upsert_trending_model_analytics(args: Args<'_>) -> AnyhowResult<()> {
+
+  let (model_type, model_token) = match args.model_token {
+    ModelToken::Tts(ref token) => (ModelType::Tts, token.as_str()),
+  };
+
+  let query = sqlx::query!(
+        r#"
+INSERT INTO trending_model_analytics
+SET
+  model_type = ?,
+  model_token = ?,
+  window_name = ?,
+  numeric_value = ?,
+  version = 1
+
+ON DUPLICATE KEY UPDATE
+  numeric_value = ?,
+  version = version + 1
+        "#,
+      // Insert
+      model_type,
+      model_token,
+      args.window_name,
+      args.numeric_value,
+      args.numeric_value,
+    );
+
+  let _r = query.execute(args.mysql_connection).await?;
+  Ok(())
+}

crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses.rs

@@ -0,0 +1,45 @@
+//! NB: This query is potentially long-lived and not suitable for low-latency HTTP endpoints.
+
+use errors::AnyhowResult;
+use log::info;
+use sqlx::MySql;
+use sqlx::pool::PoolConnection;
+use tokens::tokens::tts_models::TtsModelToken;
+
+pub struct TtsModelUseCountInfo {
+  pub use_count: u64,
+}
+
+pub async fn count_tts_model_uses(
+  tts_model_token: &TtsModelToken,
+  last_n_minutes: u64,
+  mysql_connection: &mut PoolConnection<MySql>,
+) -> AnyhowResult<TtsModelUseCountInfo> {
+
+  info!("Querying for TTS model {} use count ({} minutes)", tts_model_token, last_n_minutes);
+
+  let use_count = sqlx::query_as!(
+    RawTtsModelUseCountInfo,
+        r#"
+    SELECT
+      count(*) AS use_count
+    FROM
+      tts_results
+    WHERE
+      created_at > ( CURDATE() - INTERVAL ? MINUTE )
+      AND model_token = ?
+        "#,
+    last_n_minutes,
+    tts_model_token
+  )
+      .fetch_one(mysql_connection)
+      .await?;
+
+  Ok(TtsModelUseCountInfo {
+    use_count: use_count.use_count as u64,
+  })
+}
+
+struct RawTtsModelUseCountInfo {
+  pub use_count: i64,
+}

crates/lib/database_queries/src/queries/tts/tts_models/count_tts_model_uses_total.rs

@@ -0,0 +1,42 @@
+//! NB: This query is potentially long-lived and not suitable for low-latency HTTP endpoints.
+
+use errors::AnyhowResult;
+use log::info;
+use sqlx::MySql;
+use sqlx::pool::PoolConnection;
+use tokens::tokens::tts_models::TtsModelToken;
+
+pub struct TtsModelTotalUseCountInfo {
+  pub total_use_count: u64,
+}
+
+pub async fn count_tts_model_uses_total(
+  tts_model_token: &TtsModelToken,
+  mysql_connection: &mut PoolConnection<MySql>,
+) -> AnyhowResult<TtsModelTotalUseCountInfo> {
+
+  info!("Querying for TTS model {} total use count", tts_model_token);
+
+  let total_use_count = sqlx::query_as!(
+    RawTtsModelTotalUseCountInfo,
+        r#"
+    SELECT
+      count(*) AS total_use_count
+    FROM
+      tts_results
+    WHERE
+      model_token = ?
+        "#,
+    tts_model_token
+  )
+      .fetch_one(mysql_connection)
+      .await?;
+
+  Ok(TtsModelTotalUseCountInfo {
+    total_use_count: total_use_count.total_use_count as u64,
+  })
+}
+
+struct RawTtsModelTotalUseCountInfo {
+  pub total_use_count: i64,
+}

crates/lib/database_queries/src/queries/tts/tts_models/list_all_tts_model_tokens.rs

@@ -0,0 +1,51 @@
+use log::info;
+use errors::AnyhowResult;
+use sqlx::MySql;
+use sqlx::pool::PoolConnection;
+use tokens::tokens::tts_models::TtsModelToken;
+
+/// List of all TTS model tokens.
+/// This is typically used to run jobs that calculate analytics on models.
+/// Currently does not need to be batched, but future scale may dictate batching.
+#[derive(Clone)]
+pub struct TtsModelTokens {
+  pub tokens: Vec<TtsModelTokenInfo>,
+}
+
+#[derive(Clone)]
+pub struct TtsModelTokenInfo {
+  pub token: TtsModelToken,
+  //pub ietf_language_tag: String,
+  //pub ietf_primary_language_subtag: String,
+  //pub creator_set_visibility: Visibility,
+  //pub maybe_user_deleted_at: Option<DateTime<Utc>>,
+  //pub maybe_mod_deleted_at: Option<DateTime<Utc>>,
+}
+
+pub async fn list_all_tts_model_tokens(mysql_connection: &mut PoolConnection<MySql>) -> AnyhowResult<TtsModelTokens> {
+  info!("Querying for all TTS model tokens");
+
+  let tokens = sqlx::query_as!(
+    RawTtsModelTokenInfo,
+        r#"
+SELECT
+    token as `token: tokens::tokens::tts_models::TtsModelToken`
+FROM tts_models
+        "#)
+        .fetch_all(mysql_connection)
+        .await?;
+
+  let tokens = tokens.into_iter()
+      .map(|token| TtsModelTokenInfo {
+        token: token.token,
+      })
+      .collect::<Vec<TtsModelTokenInfo>>();
+
+  Ok(TtsModelTokens {
+    tokens,
+  })
+}
+
+struct RawTtsModelTokenInfo {
+  pub token: TtsModelToken,
+}

crates/lib/database_queries/src/queries/tts/tts_models/mod.rs

@@ -1,7 +1,10 @@
+pub mod count_tts_model_uses;
+pub mod count_tts_model_uses_total;
 pub mod delete_tts_model_various_scopes;
 pub mod edit_tts_model_details;
 pub mod edit_tts_model_moderator_details;
 pub mod get_tts_model;
 pub mod get_tts_model_for_inference;
 pub mod insert_tts_model_from_download_job;
-pub mod list_tts_models;
+pub mod list_all_tts_model_tokens;
+pub mod list_tts_models;

crates/service/job/analytics_job/Cargo.toml

@@ -0,0 +1,50 @@
+[package]
+name = "analytics-job"
+publish = false
+edition = "2018"
+version = "0.0.1"
+authors = [
+    "Brandon Thomas <bt@brand.io>",
+    "Brandon Thomas <echelon@gmail.com>",
+]
+
+[[bin]]
+name = "analytics-job"
+path = "src/main.rs"
+
+[dependencies]
+
+# Internal
+config = { path = "../../../lib/config" }
+container_common = { path = "../../../lib/container_common" }
+database_queries = { path = "../../../lib/database_queries" }
+enums = { path = "../../../_types/enums" }
+errors = { path = "../../../lib/errors" }
+jobs_common = { path = "../../../lib/jobs_common" }
+reusable_types = { path = "../../../_types/reusable_types" }
+tokens = { path = "../../../_types/tokens" }
+
+# External
+anyhow = "1.0.40"
+chrono = { version = "0.4.19", features = ["serde"] }
+clap = "2.34.0" # TODO: This is really out of date, but has conflicts when updated
+dotenv = "0.15.0"
+easyenv = "0.0.6"
+hostname = "0.3.1"
+log = "0.4.14"
+rand = "0.8.3"
+
+# 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" ] }
+
+# Runtime
+# NB: We switched to Tokio from `async-std` for broader ecosystem support.
+tokio = { version = "1.6.0", features = ["macros"] }
+
+# Serialization
+serde = "1.0.125"
+serde_derive = "1.0.125"
+serde_json = "1.0.64"

crates/service/job/analytics_job/src/job_state.rs

@@ -0,0 +1,22 @@
+use sqlx::MySqlPool;
+
+pub struct JobState {
+  pub mysql_pool: MySqlPool,
+
+  pub sleep_config: SleepConfigs,
+}
+
+/// Use sleep to not overload the database.
+pub struct SleepConfigs {
+  // How long to wait between individual "jobs".
+  pub between_job_wait_millis: u64,
+
+  // How long to wait between individual job batches.
+  pub between_job_batch_wait_millis: u64,
+
+  // How long to wait between individual queries.
+  pub between_query_wait_millis: u64,
+
+  // How long to wait between errors.
+  pub between_error_wait_millis: u64,
+}

crates/service/job/analytics_job/src/jobs/calculate_model_analytics.rs

@@ -0,0 +1,118 @@
+use crate::job_state::JobState;
+use database_queries::queries::trending_model_analytics::upsert_trending_model_analytics::{Args, ModelToken, upsert_trending_model_analytics};
+use database_queries::queries::tts::tts_models::count_tts_model_uses::count_tts_model_uses;
+use database_queries::queries::tts::tts_models::count_tts_model_uses_total::count_tts_model_uses_total;
+use database_queries::queries::tts::tts_models::list_all_tts_model_tokens::{list_all_tts_model_tokens, TtsModelTokens};
+use errors::AnyhowResult;
+use log::info;
+use rand::seq::SliceRandom;
+use rand::thread_rng;
+use sqlx::MySql;
+use sqlx::pool::PoolConnection;
+use std::time::Duration;
+use enums::by_table::trending_model_analytics::window_name::WindowName;
+use tokens::tokens::tts_models::TtsModelToken;
+
+// TODO(bt, 2023-01-16): Make a job trait and make this an instance.
+//  Jobs can store state, batch progress, and not starve one another.
+//  It'll be easier to interleave jobs.
+
+pub async fn calculate_model_analytics(job_state: &JobState) -> AnyhowResult<()> {
+  // TODO(bt, 2023-01-16): It's conceivable the pool connection dies mid-workload and we starve
+  //  our tokens. An ideal fix would be a self-healing pool connection, but for now we'll use
+  //  ugly hacks.
+  let mut mysql_connection = job_state.mysql_pool.acquire().await?;
+
+  let tokens = query_all_model_tokens(&mut mysql_connection).await?;
+
+  std::thread::sleep(Duration::from_millis(100));
+
+  for token in tokens.tokens {
+    info!("Running analytics for TTS model {}", token.token);
+
+    let _r = query_single_model_statistics(
+      job_state,
+      &mut mysql_connection,
+      &token.token).await;
+
+    std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_job_wait_millis));
+  }
+
+  Ok(())
+}
+
+async fn query_all_model_tokens(
+  mysql_connection: &mut PoolConnection<MySql>,
+) -> AnyhowResult<TtsModelTokens> {
+  let mut tokens = list_all_tts_model_tokens(mysql_connection).await?;
+
+  // NB: This is just a hack to not starve the work queue. We should replace this with a
+  // self-healing connection and batch cursors.
+  let mut rng = thread_rng();
+  tokens.tokens.shuffle(&mut rng);
+
+  Ok(tokens)
+}
+
+async fn query_single_model_statistics(
+  job_state: &JobState,
+  mysql_connection: &mut PoolConnection<MySql>,
+  model_token: &TtsModelToken,
+) -> AnyhowResult<()> {
+  {
+    let three_hours_in_minutes = 60 * 3;
+
+    let result = count_tts_model_uses(
+      model_token,
+      three_hours_in_minutes,
+      mysql_connection).await?;
+
+    info!("TTS model {} uses {} times (three hours)", model_token, result.use_count);
+
+    let _r = upsert_trending_model_analytics(Args {
+      model_token: ModelToken::Tts(model_token),
+      window_name: WindowName::Last3Hours,
+      numeric_value: result.use_count,
+      mysql_connection,
+    }).await?;
+  }
+
+  std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_query_wait_millis));
+
+  {
+    let three_days_in_minutes = 60 * 24 * 3;
+
+    let result = count_tts_model_uses(
+      model_token,
+      three_days_in_minutes,
+      mysql_connection).await?;
+
+    info!("TTS model {} uses {} times (three days)", model_token, result.use_count);
+
+    let _r = upsert_trending_model_analytics(Args {
+      model_token: ModelToken::Tts(model_token),
+      window_name: WindowName::Last3Days,
+      numeric_value: result.use_count,
+      mysql_connection,
+    }).await?;
+  }
+
+  std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_query_wait_millis));
+
+  {
+    let result = count_tts_model_uses_total(
+      model_token,
+      mysql_connection).await?;
+
+    info!("TTS model {} uses {} times (total)", model_token, result.total_use_count);
+
+    let _r = upsert_trending_model_analytics(Args {
+      model_token: ModelToken::Tts(model_token),
+      window_name: WindowName::AllTime,
+      numeric_value: result.total_use_count,
+      mysql_connection,
+    }).await?;
+  }
+
+  Ok(())
+}

crates/service/job/analytics_job/src/jobs/mod.rs

@@ -0,0 +1 @@
+pub mod calculate_model_analytics;

crates/service/job/analytics_job/src/main.rs

@@ -0,0 +1,71 @@
+// 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)]
+
+// Strict
+//#![forbid(warnings)]
+
+pub mod job_state;
+pub mod jobs;
+pub mod main_loop;
+
+use config::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING;
+use config::shared_constants::DEFAULT_RUST_LOG;
+use crate::job_state::{JobState, SleepConfigs};
+use crate::main_loop::main_loop;
+use errors::AnyhowResult;
+use log::info;
+use sqlx::mysql::MySqlPoolOptions;
+
+#[tokio::main]
+async fn main() -> AnyhowResult<()> {
+  easyenv::init_all_with_default_logging(Some(DEFAULT_RUST_LOG));
+
+  let _ = dotenv::from_filename(".env-analytics-job").ok(); // NB: Specific to `analytics-job` app.
+  let _ = dotenv::from_filename(".env-secrets").ok(); // NB: Secrets not to live in source control.
+
+  info!("Obtaining hostname...");
+
+  let server_hostname = hostname::get()
+      .ok()
+      .and_then(|h| h.into_string().ok())
+      .unwrap_or("analytics-job".to_string());
+
+  info!("Hostname: {}", &server_hostname);
+
+  let db_connection_string =
+      easyenv::get_env_string_or_default(
+        "MYSQL_URL",
+        DEFAULT_MYSQL_CONNECTION_STRING);
+
+  info!("Connecting to database...");
+
+  let mysql_pool = MySqlPoolOptions::new()
+      .max_connections(5)
+      .connect(&db_connection_string)
+      .await?;
+
+  let job_state = JobState {
+    mysql_pool,
+    sleep_config: SleepConfigs {
+      between_job_wait_millis: easyenv::get_env_num("BETWEEN_JOB_WAIT_MILLIS", 100)?,
+      between_job_batch_wait_millis: easyenv::get_env_num("BETWEEN_JOB_BATCH_WAIT_MILLIS", 5000)?,
+      between_query_wait_millis: easyenv::get_env_num("BETWEEN_QUERY_WAIT_MILLIS", 100)?,
+      between_error_wait_millis: easyenv::get_env_num("BETWEEN_ERROR_WAIT_MILLIS", 10_000)?,
+    },
+  };
+
+  let _r = main_loop(job_state).await;
+
+  Ok(())
+}

crates/service/job/analytics_job/src/main_loop.rs

@@ -0,0 +1,21 @@
+use crate::job_state::JobState;
+use crate::jobs::calculate_model_analytics::calculate_model_analytics;
+use errors::AnyhowResult;
+use std::time::Duration;
+use log::{error, info};
+
+pub async fn main_loop(job_state: JobState) -> AnyhowResult<()> {
+  loop {
+    info!("Begin analytics job batch.");
+
+    match calculate_model_analytics(&job_state).await {
+      Ok(_) => {
+        std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_job_batch_wait_millis));
+      }
+      Err(e) => {
+        error!("Error: {:?}", e);
+        std::thread::sleep(Duration::from_millis(job_state.sleep_config.between_error_wait_millis));
+      }
+    }
+  }
+}

crates/service/web/storyteller_web/src/http_server/endpoints/categories/tts/list_fully_computed_assigned_tts_categories.rs

@@ -26,9 +26,10 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fmt;
 use std::sync::Arc;
 use chrono::{DateTime, Utc};
+use database_queries::queries::trending_model_analytics::list_trending_tts_models::{list_trending_tts_models, TrendingModels};
 use tokens::tokens::model_categories::ModelCategoryToken;
 use tokens::tokens::tts_models::TtsModelToken;
-use crate::model::categories::synthetic_category_list::SYNTHETIC_CATEGORY_LATEST_TTS_MODELS;
+use crate::model::categories::synthetic_category_list::{SYNTHETIC_CATEGORY_LATEST_TTS_MODELS, SYNTHETIC_CATEGORY_TRENDING_TTS_MODELS};
 
 // =============== Success Response ===============
 
@@ -177,7 +178,12 @@ async fn query_and_construct_payload(
   category_cache: &SingleItemTtlCache<CategoryList>,
   mysql_pool: &MySqlPool
 ) -> Result<ModelTokensByCategoryToken, ListFullyComputedAssignedTtsCategoriesError> {
-  let (categories, models, model_category_map) = {
+  let (
+    categories,
+    models,
+    model_category_map,
+    trending_models
+  ) = {
     let mut mysql_connection = mysql_pool.acquire()
         .await
         .map_err(|e| {
@@ -206,12 +212,20 @@ async fn query_and_construct_payload(
           ListFullyComputedAssignedTtsCategoriesError::ServerError
         })?;
 
-    (categories, models, model_category_map)
+    let trending_models = list_trending_tts_models(&mut mysql_connection)
+        .await
+        .map_err(|e| {
+          error!("Error querying trending TTS models: {:?}", e);
+          ListFullyComputedAssignedTtsCategoriesError::ServerError
+        })?;
+
+    (categories, models, model_category_map, trending_models)
   };
 
   let mut recursive_category_to_model_map = recursive_category_to_model_map(&model_category_map, &categories);
 
   add_recent_models(&mut recursive_category_to_model_map, models);
+  add_trending_models(&mut recursive_category_to_model_map, trending_models);
 
   Ok(ModelTokensByCategoryToken {
     recursive: recursive_category_to_model_map,
@@ -435,6 +449,19 @@ fn add_recent_models(recursive_category_to_model_map: &mut CategoryTokenToModelT
   recursive_category_to_model_map.insert(synthetic_token, model_tokens);
 }
 
+fn add_trending_models(recursive_category_to_model_map: &mut CategoryTokenToModelTokensMap, trending_models: TrendingModels) {
+  // TODO: This is a grossly bad guess.
+  let model_tokens = trending_models.models.iter()
+      .take(30)
+      .map(|trending_model| {
+        trending_model.model_token.clone()
+      })
+      .collect::<BTreeSet<TtsModelToken>>();
+
+  let synthetic_token = ModelCategoryToken::new_from_str(SYNTHETIC_CATEGORY_TRENDING_TTS_MODELS.category_token);
+  recursive_category_to_model_map.insert(synthetic_token, model_tokens);
+}
+
 #[cfg(test)]
 mod tests {
   // TODO: This really needs tests!

crates/service/web/storyteller_web/src/http_server/endpoints/mod.rs

@@ -10,6 +10,7 @@ pub mod misc;
 pub mod moderation;
 pub mod service;
 pub mod stubs;
+pub mod trending;
 pub mod tts;
 pub mod twitch;
 pub mod vocoders;

crates/service/web/storyteller_web/src/http_server/endpoints/trending/list_trending_tts_models.rs

@@ -0,0 +1,152 @@
+// NB: Incrementally getting rid of build warnings...
+//#![forbid(unused_imports)]
+//#![forbid(unused_mut)]
+//#![forbid(unused_variables)]
+
+//! This endpoint recursively calculates (and caches) the list of every category a TTS model
+//! belongs to. This saves an enormous amount of clientside CPU compute.
+//!
+
+use actix_web::error::ResponseError;
+use actix_web::http::StatusCode;
+use actix_web::{web, HttpResponse, HttpRequest};
+use chrono::{DateTime, Utc};
+use container_common::anyhow_result::AnyhowResult;
+use crate::model::cached_queries::list_cached_tts_categories_for_public_dropdown::list_cached_tts_categories_for_public_dropdown;
+use crate::model::categories::synthetic_category_list::SYNTHETIC_CATEGORY_LATEST_TTS_MODELS;
+use crate::server_state::ServerState;
+use database_queries::queries::model_categories::list_categories_query_builder::CategoryList;
+use database_queries::queries::trending_model_analytics::list_trending_tts_models::{list_trending_tts_models, list_trending_tts_models_with_pool};
+use database_queries::queries::tts::tts_category_assignments::fetch_and_build_tts_model_category_map::fetch_and_build_tts_model_category_map_with_connection;
+use database_queries::queries::tts::tts_models::list_tts_models::list_tts_models_with_connection;
+use enums::by_table::trending_model_analytics::window_name::WindowName;
+use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
+use log::{info, error};
+use memory_caching::single_item_ttl_cache::SingleItemTtlCache;
+use sqlx::pool::PoolConnection;
+use sqlx::{MySql, MySqlPool};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+use tokens::tokens::model_categories::ModelCategoryToken;
+use tokens::tokens::tts_models::TtsModelToken;
+
+// TODO TODO TODO: This endpoint is not done!
+// TODO TODO TODO: This endpoint is not done!
+// TODO TODO TODO: This endpoint is not done!
+// TODO TODO TODO: This endpoint is not done!
+// TODO TODO TODO: This endpoint is not done!
+// TODO TODO TODO: This endpoint is not done!
+
+// =============== Success Response ===============
+
+#[derive(Serialize)]
+pub struct ListTrendingTtsModelsResponse {
+  pub success: bool,
+  
+  pub top_trending: WindowTrends,
+
+  pub top_trending_by_language_code: HashMap<String, WindowTrends>,
+}
+
+pub type WindowTrends = HashMap<WindowName, Vec<TtsModelToken>>;
+
+
+// =============== Error Response ===============
+
+#[derive(Debug, Serialize)]
+pub enum ListTrendingTtsModelsError {
+  ServerError,
+}
+
+impl ResponseError for ListTrendingTtsModelsError {
+  fn status_code(&self) -> StatusCode {
+    match *self {
+      ListTrendingTtsModelsError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
+    }
+  }
+
+  fn error_response(&self) -> HttpResponse {
+    serialize_as_json_error(self)
+  }
+}
+
+// NB: Not using derive_more::Display since Clion doesn't understand it.
+impl fmt::Display for ListTrendingTtsModelsError {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    write!(f, "{:?}", self)
+  }
+}
+
+// =============== Handler ===============
+
+pub async fn list_trending_tts_models_handler(
+  _http_request: HttpRequest,
+  server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, ListTrendingTtsModelsError>
+{
+
+// TODO: Cache the outputs!
+
+//  let maybe_category_assignments = server_state.caches.tts_model_category_assignments.copy_without_bump_if_unexpired()
+//      .map_err(|e| {
+//        error!("Error consulting cache: {:?}", e);
+//        ListTrendingTtsModelsError::ServerError
+//      })?;
+//
+//  let category_assignments = match maybe_category_assignments {
+//    Some(category_assignments) => {
+//      info!("Serving TTS category assignments from cache");
+//      category_assignments
+//    },
+//    None => {
+//      let category_assignments = query_and_construct_payload(
+//        &server_state.caches.database_tts_category_list,
+//        &server_state.mysql_pool)
+//          .await?;
+//
+//      server_state.caches.tts_model_category_assignments.store_copy(&category_assignments)
+//          .map_err(|e| {
+//            error!("Error storing cache: {:?}", e);
+//            ListTrendingTtsModelsError::ServerError
+//          })?;
+//
+//      category_assignments
+//    },
+//  };
+
+  let trending_models= list_trending_tts_models_with_pool(&server_state.mysql_pool).await
+      .map_err(|e| {
+        error!("Query error: {:?}", e);
+        ListTrendingTtsModelsError::ServerError
+      })?;
+
+// TODO: Actually generate the response body sensibly.
+
+  let mut top_trending = HashMap::new();
+  let mut top_trending_by_language_code = HashMap::new();
+
+  for trending_model in trending_models.models.iter() {
+    if !top_trending_by_language_code.contains_key(&trending_model.ietf_primary_language_subtag) {
+      top_trending_by_language_code.insert(trending_model.ietf_primary_language_subtag.clone(), HashMap::new());
+    }
+
+    if let Some(window_map) = top_trending_by_language_code.get_mut(&trending_model.ietf_primary_language_subtag) {
+      if !window_map.contains_key(&trending_model.window_name) {
+        window_map.insert(trending_model.window_name, Vec::new());
+      }
+    }
+  }
+
+  let response = ListTrendingTtsModelsResponse {
+    success: true,
+    top_trending,
+    top_trending_by_language_code,
+  };
+
+  let body = serde_json::to_string(&response)
+      .map_err(|_e| ListTrendingTtsModelsError::ServerError)?;
+
+  Ok(HttpResponse::Ok()
+      .content_type("application/json")
+      .body(body))
+}

crates/service/web/storyteller_web/src/http_server/endpoints/trending/mod.rs

@@ -0,0 +1 @@
+pub mod list_trending_tts_models;

crates/service/web/storyteller_web/src/routes.rs

@@ -52,6 +52,7 @@ use crate::http_server::endpoints::stubs::app_model_downloads::get_app_model_dow
 use crate::http_server::endpoints::stubs::app_news::get_app_news_handler;
 use crate::http_server::endpoints::stubs::app_plans::get_app_plans_handler;
 use crate::http_server::endpoints::stubs::post_app_analytics::post_app_analytics_handler;
+use crate::http_server::endpoints::trending::list_trending_tts_models::list_trending_tts_models_handler;
 use crate::http_server::endpoints::tts::delete_tts_model::delete_tts_model_handler;
 use crate::http_server::endpoints::tts::delete_tts_result::delete_tts_inference_result_handler;
 use crate::http_server::endpoints::tts::edit_tts_model::edit_tts_model_handler;
@@ -128,6 +129,7 @@ pub fn add_routes<T, B> (app: App<T, B>) -> App<T, B>
   app = add_flag_routes(app); /* /flag */
   app = add_desktop_app_routes(app); /* /v1/vc/... */
   app = add_media_upload_routes(app); /* /v1/media_upload/... */
+  app = add_trending_routes(app); /* /v1/trending/... */
 
   // From components
   app = add_suggested_api_v1_account_creation_and_session_routes(app); // /create_account, /session, /login, /logout
@@ -905,6 +907,28 @@ fn add_media_upload_routes<T, B> (app: App<T, B>) -> App<T, B>
   )
 }
 
+// ==================== TRENDING ROUTES ====================
+
+fn add_trending_routes<T, B> (app: App<T, B>) -> App<T, B>
+  where
+      B: MessageBody,
+      T: ServiceFactory<
+        ServiceRequest,
+        Config = (),
+        Response = ServiceResponse<B>,
+        Error = Error,
+        InitError = (),
+      >,
+{
+  app.service(web::scope("/v1/trending")
+      .service(web::resource("/tts_models")
+          .route(web::get().to(list_trending_tts_models_handler))
+          .route(web::head().to(|| HttpResponse::Ok()))
+      )
+  )
+}
+
+
 // ==================== TWITCH ROUTES ====================
 
 // TODO: Maybe move these into an "oauth-gateway" type http service.