Add in-memory cache for trending queries

rsky-feedgen/Cargo.toml

@@ -27,6 +27,7 @@ regex = "1.8.4"
 base64 = "0.21.2"
 rand = "0.8.5"
 once_cell = "1.19.0"
+moka = { version = "0.12", features = ["future"] }
 
 [dependencies.rocket_sync_db_pools]
 version = "=0.1.0"

rsky-feedgen/src/apis/mod.rs

@@ -9,6 +9,7 @@ use diesel::dsl::sql;
 use diesel::prelude::*;
 use diesel::sql_query;
 use diesel::sql_types::{Array, Bool, Nullable, Text};
+use moka::future::Cache;
 use once_cell::sync::Lazy;
 use rand::Rng;
 use regex::Regex;
@@ -18,7 +19,7 @@ use rsky_common::explicit_slurs::contains_explicit_slurs;
 use rsky_lexicon::app::bsky::embed::{Embeds, MediaUnion};
 use rsky_lexicon::app::bsky::feed::PostLabels;
 use std::collections::HashSet;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 #[allow(deprecated)]
 pub async fn get_posts_by_membership(
@@ -157,6 +158,7 @@ pub async fn get_posts_by_membership(
     result
 }
 
+#[derive(Clone)]
 pub enum TrendingMedia {
     OnlyVideo,
     OnlyImage,
@@ -301,6 +303,56 @@ pub async fn get_blacksky_trending(
     result
 }
 
+// Global cache using Moka. Keys are strings (derived from query parameters) and
+// values are AlgoResponse.
+static TRENDING_CACHE: Lazy<Cache<String, AlgoResponse>> = Lazy::new(|| {
+    Cache::builder()
+        .max_capacity(env_int("FEEDGEN_TRENDING_CACHE_CAPACITY").unwrap_or(100) as u64)
+        .time_to_live(Duration::from_secs(
+            env_int("FEEDGEN_TRENDING_CACHE_TTL").unwrap_or(2) as u64,
+        ))
+        .build()
+});
+
+// Build a cache key from the parameters.
+fn build_trending_cache_key(
+    limit: Option<i64>,
+    params_cursor: Option<&str>,
+    media: Option<TrendingMedia>,
+) -> String {
+    let media_str = match media {
+        Some(TrendingMedia::OnlyImage) => "OnlyImage",
+        Some(TrendingMedia::OnlyVideo) => "OnlyVideo",
+        None => "None",
+    };
+    format!(
+        "limit:{}|cursor:{}|media:{}",
+        limit.unwrap_or(30),
+        params_cursor.unwrap_or(""),
+        media_str
+    )
+}
+
+// Cached version of get_blacksky_trending.
+pub async fn get_blacksky_trending_cached(
+    limit: Option<i64>,
+    params_cursor: Option<&str>,
+    connection: ReadReplicaConn,
+    config: &FeedGenConfig,
+    media: Option<TrendingMedia>,
+) -> Result<AlgoResponse, ValidationErrorMessageResponse> {
+    let cache_key = build_trending_cache_key(limit, params_cursor, media.clone());
+    // Try to retrieve the result from the cache.
+    if let Some(cached_response) = TRENDING_CACHE.get(&cache_key).await {
+        return Ok(cached_response);
+    }
+    // If not found, compute the result.
+    let result = get_blacksky_trending(limit, params_cursor, connection, config, media).await?;
+    // Insert the computed result into the cache.
+    TRENDING_CACHE.insert(cache_key, result.clone()).await;
+    Ok(result)
+}
+
 #[allow(deprecated)]
 pub async fn get_all_posts(
     lang: Option<String>,

rsky-feedgen/src/routes.rs

@@ -189,7 +189,8 @@ pub async fn index(
             }
         }
         _blacksky_trend if _blacksky_trend == BLACKSKY_TREND && !is_banned => {
-            match crate::apis::get_blacksky_trending(limit, cursor, connection, config, None).await
+            match crate::apis::get_blacksky_trending_cached(limit, cursor, connection, config, None)
+                .await
             {
                 Ok(response) => Ok(Json(response)),
                 Err(error) => {
@@ -206,7 +207,7 @@ pub async fn index(
             }
         }
         _blacksky_videos if _blacksky_videos == BLACKSKY_VIDEOS && !is_banned => {
-            match crate::apis::get_blacksky_trending(
+            match crate::apis::get_blacksky_trending_cached(
                 limit,
                 cursor,
                 connection,
@@ -230,7 +231,7 @@ pub async fn index(
             }
         }
         _blacksky_photos if _blacksky_photos == BLACKSKY_PHOTOS && !is_banned => {
-            match crate::apis::get_blacksky_trending(
+            match crate::apis::get_blacksky_trending_cached(
                 limit,
                 cursor,
                 connection,