~nickbp/force-rss: Implement caching, with a bit of cleanup

5 files changed, 137 insertions(+), 65 deletions(-)

M src/http.rs
M src/logging.rs
M src/main.rs
M src/rss.rs
M src/twitch.rs

M src/http.rs => src/http.rs +75 -28

@@ 2,6 2,7 @@ use anyhow::{Context, Result};
 use async_lock::Mutex;
 use async_std::task;
 use std::sync::Arc;
+use std::time::Duration;
 use tide::Request;
 use tracing::{debug, error, info, warn};
 


@@ 11,8 12,11 @@ struct State {
     twitch_auth: twitch::TwitchAuth,
     thumbnail_width: String,
     thumbnail_height: String,
-    user_cache: retainer::cache::Cache<String, usize>,
-    video_cache: retainer::cache::Cache<usize, Vec<twitch::GetVideosEntry>>,
+    user_cache_secs: u64,
+    video_cache_secs: u64,
+    username_to_user_cache: Arc<retainer::cache::Cache<String, twitch::GetUsersEntry>>,
+    userid_to_videos_cache: Arc<retainer::cache::Cache<String, Vec<twitch::GetVideosEntry>>>,
+    _cache_monitors: Vec<async_std::task::JoinHandle<()>>,
 }
 
 pub fn run(


@@ 20,21 24,42 @@ pub fn run(
     twitch_auth: twitch::TwitchAuth,
     thumbnail_width: String,
     thumbnail_height: String,
+    user_cache_secs: u64,
+    video_cache_secs: u64,
 ) -> Result<()> {
-    let state = State {
+    let mut state = State {
         twitch_auth,
         thumbnail_width,
         thumbnail_height,
-        user_cache: retainer::cache::Cache::new(),
-        video_cache: retainer::cache::Cache::new(),
+        user_cache_secs,
+        video_cache_secs,
+        username_to_user_cache: Arc::new(retainer::cache::Cache::new()),
+        userid_to_videos_cache: Arc::new(retainer::cache::Cache::new()),
+        _cache_monitors: vec![],
     };
+    let user_cache_clone = state.username_to_user_cache.clone();
+    state._cache_monitors.push(task::spawn(async move {
+        // Every 3s, check 4 entries for expiration.
+        // If >25% were expired, then repeat the check for another 4 entries
+        user_cache_clone
+            .monitor(4, 0.25, Duration::from_secs(3))
+            .await
+    }));
+    let video_cache_clone = state.userid_to_videos_cache.clone();
+    state._cache_monitors.push(task::spawn(async move {
+        // Every 3s, check 4 entries for expiration.
+        // If >25% were expired, then repeat the check for another 4 entries
+        video_cache_clone
+            .monitor(4, 0.25, Duration::from_secs(3))
+            .await
+    }));
 
     let mut app = tide::with_state(Arc::new(Mutex::new(state)));
     // Enable request logging if debug (or trace)
     app.with(LogMiddleware {});
     app.at("/").get(handle_index);
     app.at("/rss").get(handle_rss);
-    info!("Listening at {}", listen_endpoint);
+    info!("Listening at LISTEN={}", listen_endpoint);
     return task::block_on(app.listen(listen_endpoint)).context("HTTP listener exited");
 }
 


@@ 66,33 91,55 @@ async fn handle_rss(req: Request<Arc<Mutex<State>>>) -> tide::Result {
 
     match query_account {
         Some(account) => {
-            // TODO check userid cache, then try get_user_id
-            // TODO check video cache, then try get_videos
             let response_body: String;
             {
                 let mut state = req.state().lock().await;
                 let user: twitch::GetUsersEntry;
                 let videos: Vec<twitch::GetVideosEntry>;
-                match twitch::get_user(&mut state.twitch_auth, account).await {
-                    Ok(u) => {
-                        user = u;
-                    }
-                    Err(e) => {
-                        return Ok(tide::Response::builder(400)
-                            .body(format!("400 Bad Request: {}", e))
-                            .content_type(tide::http::mime::PLAIN)
-                            .build());
-                    }
-                };
-                match twitch::get_videos(&mut state.twitch_auth, &user.id).await {
-                    Ok(v) => {
-                        videos = v;
-                    }
-                    Err(e) => {
-                        return Ok(tide::Response::builder(400)
-                            .body(format!("400 Bad Request: {}", e))
-                            .content_type(tide::http::mime::PLAIN)
-                            .build());
+                if let Some(u) = state.username_to_user_cache.get(&account).await {
+                    user = (*u).clone();
+                } else {
+                    match twitch::get_user(&mut state.twitch_auth, &account).await {
+                        Ok(u) => {
+                            state
+                                .username_to_user_cache
+                                .insert(
+                                    account,
+                                    u.clone(),
+                                    Duration::from_secs(state.user_cache_secs),
+                                )
+                                .await;
+                            user = u;
+                        }
+                        Err(e) => {
+                            return Ok(tide::Response::builder(400)
+                                .body(format!("400 Bad Request: {}", e))
+                                .content_type(tide::http::mime::PLAIN)
+                                .build());
+                        }
+                    };
+                }
+                if let Some(v) = state.userid_to_videos_cache.get(&user.id).await {
+                    videos = (*v).clone();
+                } else {
+                    match twitch::get_videos(&mut state.twitch_auth, &user.id).await {
+                        Ok(v) => {
+                            state
+                                .userid_to_videos_cache
+                                .insert(
+                                    user.id.clone(),
+                                    v.clone(),
+                                    Duration::from_secs(state.video_cache_secs),
+                                )
+                                .await;
+                            videos = v;
+                        }
+                        Err(e) => {
+                            return Ok(tide::Response::builder(400)
+                                .body(format!("400 Bad Request: {}", e))
+                                .content_type(tide::http::mime::PLAIN)
+                                .build());
+                        }
                     }
                 };
                 response_body = rss::to_rss(

M src/logging.rs => src/logging.rs +5 -5

@@ 1,6 1,5 @@
 use tracing;
-use tracing_subscriber::layer::SubscriberExt;
-use tracing_subscriber::{fmt, EnvFilter};
+use tracing_subscriber::EnvFilter;
 
 pub fn init_logging() {
     let filter_layer = EnvFilter::try_from_env("LOG_LEVEL")


@@ 8,9 7,10 @@ pub fn init_logging() {
         .expect("Failed to initialize filter layer");
 
     tracing::subscriber::set_global_default(
-        tracing_subscriber::Registry::default()
-            .with(filter_layer)
-            .with(fmt::layer()),
+        tracing_subscriber::fmt()
+            .with_writer(std::io::stderr)
+            .with_env_filter(filter_layer)
+            .finish(),
     )
     .expect("Failed to set default subscriber");
 }

M src/main.rs => src/main.rs +26 -9

@@ 7,7 7,7 @@ use anyhow::{bail, Context, Result};
 use async_std::task;
 use git_testament::{git_testament, render_testament};
 use std::env;
-use tracing::{debug, error, info};
+use tracing::{error, info};
 
 use crate::twitch::TwitchAuth;
 


@@ 34,9 34,12 @@ fn env_help() -> &'static str {
 Environment variables:
   CLIENT_ID:     Twitch Client ID for querying feeds
   CLIENT_SECRET: Twitch Client Secret for querying feeds
-  LISTEN:        Endpoint for HTTP service to listen on
-                 (default: '0.0.0.0:8080')
-  LOG_LEVEL:     error/warn/info/debug/trace/off";
+  LOG_LEVEL:     error/warn/info/debug/trace/off
+
+HTTP mode only:
+  LISTEN:           Endpoint for HTTP service to listen on (default: '0.0.0.0:8080')
+  USER_CACHE_SECS:  Duration in seconds to cache Twitch account info (default: 86400)
+  VIDEO_CACHE_SECS: Duration in seconds to cache video lists (default: 600)";
 }
 
 fn main() -> Result<()> {


@@ 56,7 59,7 @@ fn main() -> Result<()> {
             task::block_on(async move {
                 let user: twitch::GetUsersEntry;
                 let videos: Vec<twitch::GetVideosEntry>;
-                match twitch::get_user(&mut twitch_auth, username.clone()).await {
+                match twitch::get_user(&mut twitch_auth, &username).await {
                     Ok(u) => {
                         user = u;
                     }


@@ 65,7 68,6 @@ fn main() -> Result<()> {
                         return;
                     }
                 };
-                debug!("{:?}", user);
                 match twitch::get_videos(&mut twitch_auth, &user.id).await {
                     Ok(v) => {
                         videos = v;


@@ 78,9 80,18 @@ fn main() -> Result<()> {
                         return;
                     }
                 };
-                debug!("{:?}", videos);
-                println!("{:?}", videos);
-                // TODO rss to stdout
+                match rss::to_rss(
+                    &user,
+                    &videos,
+                    &optenv("THUMBNAIL_WIDTH", "640"),
+                    &optenv("THUMBNAIL_HEIGHT", "360"),
+                ) {
+                    Ok(content) => println!("{}", content),
+                    Err(e) => {
+                        error!("Failed to render RSS: {}", e);
+                        return;
+                    }
+                };
             });
             Ok(())
         }


@@ 94,6 105,12 @@ fn main() -> Result<()> {
                 ),
                 optenv("THUMBNAIL_WIDTH", "640"),
                 optenv("THUMBNAIL_HEIGHT", "360"),
+                optenv("USER_CACHE_SECS", "86400")
+                    .parse::<u64>()
+                    .with_context(|| "USER_CACHE_SECS must be an unsigned int")?,
+                optenv("VIDEO_CACHE_SECS", "600")
+                    .parse::<u64>()
+                    .with_context(|| "VIDEO_CACHE_SECS must be an unsigned int")?,
             )
         }
     }

M src/rss.rs => src/rss.rs +27 -19

@@ 11,26 11,31 @@ pub fn to_rss(
 ) -> Result<String> {
     let mut items = vec![];
     for video in videos {
+        let mut item_builder = ItemBuilder::default();
+        item_builder
+            .title(format!("{} ({})", video.title, video.duration))
+            .content(format!(
+                "<p><img src='{}'/></p><p>{}</p>",
+                video
+                    .thumbnail_url
+                    .replace("%{width}", thumbnail_width)
+                    .replace("%{height}", thumbnail_height),
+                video.duration
+            ))
+            .link(video.url.to_string())
+            .pub_date(video.published_at.to_string());
+        if !video.description.is_empty() {
+            item_builder.description(video.description.to_string());
+        }
         items.push(
-            ItemBuilder::default()
-                .title(video.title.clone())
-                .content(format!(
-                    "<p><img src='{}'/></p><p>{}</p>",
-                    video
-                        .thumbnail_url
-                        .replace("%{width}", thumbnail_width)
-                        .replace("%{height}", thumbnail_height),
-                    video.duration
-                ))
-                .link(video.url.clone())
-                .description(format!("{:?}", video)) // TODO TEMP
-                .pub_date(video.published_at.clone())
+            item_builder
                 .build()
                 .map_err(|e| anyhow!("Error when rendering RSS item: {}", e))?,
         );
     }
-    let channel = ChannelBuilder::default()
-        .title(format!("Twitch: {}", user.display_name))
+    let mut channel_builder = ChannelBuilder::default();
+    channel_builder
+        .title(user.display_name.to_string())
         .image(Some(
             ImageBuilder::default()
                 .url(user.profile_image_url.as_str())


@@ 38,10 43,13 @@ pub fn to_rss(
                 .map_err(|e| anyhow!("Error when rendering RSS channel image: {}", e))?,
         ))
         .link(format!("https://twitch.tv/{}", user.login))
-        .description(user.description.as_str())
         .generator("twitch-rss".to_string())
-        .items(items)
+        .items(items);
+    if !user.description.is_empty() {
+        channel_builder.description(user.description.as_str());
+    }
+    Ok(channel_builder
         .build()
-        .map_err(|e| anyhow!("Error when rendering RSS channel: {}", e))?;
-    Ok(channel.to_string())
+        .map_err(|e| anyhow!("Error when rendering RSS channel: {}", e))?
+        .to_string())
 }

M src/twitch.rs => src/twitch.rs +4 -4

@@ 49,7 49,7 @@ struct GetUsersResponse {
 "view_count":102251,
 "created_at":"2021-05-20T22:34:07.072555Z"
 */
-#[derive(Debug, Deserialize)]
+#[derive(Clone, Debug, Deserialize)]
 pub struct GetUsersEntry {
     pub id: String,
     pub login: String,


@@ 89,7 89,7 @@ struct GetVideosResponse {
 "duration": "3h11m2s",
 "muted_segments": null
 */
-#[derive(Debug, Deserialize)]
+#[derive(Clone, Debug, Deserialize)]
 pub struct GetVideosEntry {
     pub id: String,
     pub title: String,


@@ 196,9 196,9 @@ impl TwitchAuth {
 
 // see also https://dev.twitch.tv/docs/api/reference#get-users
 // curl -v -H 'Client-Id: CLIENT_ID' -H 'Authorization: Bearer AUTH_TOKEN' "https://api.twitch.tv/helix/users?login=USER_NAME"
-pub async fn get_user(auth: &mut TwitchAuth, user_name: String) -> Result<GetUsersEntry> {
+pub async fn get_user(auth: &mut TwitchAuth, user_name: &str) -> Result<GetUsersEntry> {
     let query = GetUsersQuery {
-        login: user_name.clone(),
+        login: user_name.to_string(),
     };
     let mut response = send(helix_request("users", &query, auth).await?).await?;
     if !response.status().is_success() {