Timed comments on the waveform

Each comment sits as a dot under the moment it was left at. When the music
gets there the dot pings and a bubble grows out of it on a spark, then drifts
up and fades; a mouse over a dot peeks at it. The server fetches the newest
60 comments from pl8list on demand and keeps them for 15 minutes. A setting
turns them off.

backend/src/comments.rs

@@ -0,0 +1,154 @@
+//! Timed comments of a track, for the waveform. pl8list pins every comment to a
+//! moment of the track (`timestampSeconds`) and serves them newest first, 20 to a
+//! page. We fetch a few pages on demand and keep them in memory for a while, so a
+//! track that comes round again costs pl8list nothing.
+
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+
+use serde::Serialize;
+use serde_json::Value;
+use tokio::time::Instant;
+
+use crate::pl8::Pl8;
+
+/// Pages fetched per track: the newest 60 comments.
+const PAGES: i64 = 3;
+const TTL: Duration = Duration::from_secs(15 * 60);
+/// Tracks kept; the oldest entry goes when a new one comes in.
+const KEEP: usize = 512;
+/// Longest comment body shown, in characters.
+const MAX_BODY: usize = 200;
+
+#[derive(Clone, Serialize, Debug, PartialEq)]
+pub struct Comment {
+    pub id: i64,
+    /// Seconds into the track.
+    pub at: f64,
+    pub body: String,
+    pub handle: String,
+    pub name: String,
+    pub img: Option<String>,
+    pub verified: bool,
+}
+
+/// When a track's comments were fetched, and the comments.
+type Entry = (Instant, Arc<Vec<Comment>>);
+
+#[derive(Default)]
+pub struct Comments {
+    map: Mutex<HashMap<i64, Entry>>,
+}
+
+impl Comments {
+    pub fn cached(&self, track_id: i64) -> Option<Arc<Vec<Comment>>> {
+        let map = self.map.lock().unwrap();
+        map.get(&track_id).filter(|(at, _)| at.elapsed() < TTL).map(|(_, c)| c.clone())
+    }
+
+    /// Fetch the comments from pl8list and keep them. A failed page ends the
+    /// fetch with what came before it; a failed first page is an error.
+    pub async fn fetch(&self, pl8: &Pl8, track_id: i64, handle: &str, slug: &str) -> anyhow::Result<Arc<Vec<Comment>>> {
+        let mut out = vec![];
+        for page in 1..=PAGES {
+            let res = pl8.comments(handle, slug, page).await;
+            let data = match res {
+                Ok(Some(d)) => d,
+                Ok(None) => break,
+                Err(e) if page == 1 => return Err(e),
+                Err(_) => break,
+            };
+            out.extend(parse(&data));
+            let total = data.get("totalPages").and_then(Value::as_i64).unwrap_or(1);
+            if page >= total {
+                break;
+            }
+        }
+        out.sort_by(|a, b| a.at.total_cmp(&b.at));
+        let out = Arc::new(out);
+        let mut map = self.map.lock().unwrap();
+        if map.len() >= KEEP && !map.contains_key(&track_id) {
+            map.retain(|_, (at, _)| at.elapsed() < TTL);
+            if map.len() >= KEEP
+                && let Some(oldest) = map.iter().min_by_key(|(_, (at, _))| *at).map(|(k, _)| *k)
+            {
+                map.remove(&oldest);
+            }
+        }
+        map.insert(track_id, (Instant::now(), out.clone()));
+        Ok(out)
+    }
+}
+
+/// Comments from one page of `/api/{handle}/{slug}/comments`. Deleted, hidden,
+/// untimed and empty ones are left out.
+pub fn parse(data: &Value) -> Vec<Comment> {
+    let Some(list) = data.get("comments").and_then(Value::as_array) else { return vec![] };
+    list.iter()
+        .filter_map(|c| {
+            let flag = |k: &str| c.get(k).and_then(Value::as_bool).unwrap_or(false);
+            if flag("isDeleted") || flag("isHiddenByOwner") {
+                return None;
+            }
+            let at = c.get("timestampSeconds")?.as_f64().filter(|t| t.is_finite() && *t >= 0.0)?;
+            let body = clean(c.get("body")?.as_str()?);
+            if body.is_empty() {
+                return None;
+            }
+            let user = c.get("user")?;
+            let handle = user.get("handle")?.as_str()?.to_string();
+            let name = user.get("name").and_then(Value::as_str).filter(|n| !n.trim().is_empty()).unwrap_or(&handle).trim().to_string();
+            Some(Comment {
+                id: c.get("id")?.as_i64()?,
+                at,
+                body,
+                name,
+                img: user.get("img").and_then(Value::as_str).filter(|s| !s.is_empty()).map(String::from),
+                verified: user.get("isVerified").and_then(Value::as_bool).unwrap_or(false),
+                handle,
+            })
+        })
+        .collect()
+}
+
+/// One line, at most `MAX_BODY` characters.
+fn clean(body: &str) -> String {
+    let one_line = body.split_whitespace().collect::<Vec<_>>().join(" ");
+    match one_line.char_indices().nth(MAX_BODY) {
+        Some((cut, _)) => format!("{}…", one_line[..cut].trim_end()),
+        None => one_line,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn parse_keeps_timed_visible_comments() {
+        let d = json!({"comments": [
+            {"id": 1, "body": "  gass\n\nat 145  ", "timestampSeconds": 37.5, "isDeleted": false,
+             "user": {"handle": "a", "name": "", "img": "x-1", "isVerified": true}},
+            {"id": 2, "body": "gone", "timestampSeconds": 3.0, "isDeleted": true, "user": {"handle": "b"}},
+            {"id": 3, "body": "hidden", "timestampSeconds": 3.0, "isHiddenByOwner": true, "user": {"handle": "b"}},
+            {"id": 4, "body": "untimed", "timestampSeconds": null, "user": {"handle": "b"}},
+            {"id": 5, "body": "   ", "timestampSeconds": 1.0, "user": {"handle": "b"}},
+        ]});
+        let c = parse(&d);
+        assert_eq!(c.len(), 1);
+        assert_eq!((c[0].id, c[0].at, c[0].body.as_str()), (1, 37.5, "gass at 145"));
+        assert_eq!((c[0].name.as_str(), c[0].img.as_deref(), c[0].verified), ("a", Some("x-1"), true));
+    }
+
+    #[test]
+    fn long_bodies_are_cut() {
+        let long = "ä".repeat(MAX_BODY + 10);
+        let cut = clean(&long);
+        assert_eq!(cut.chars().count(), MAX_BODY + 1);
+        assert!(cut.ends_with('…'));
+    }
+}

backend/src/http/app.rs

@@ -183,6 +183,38 @@ pub async fn track(State(app): St, Extension(current): Extension<CurrentUser>, Q
     Ok((no_store(), Json(json!({ "track": track }))).into_response())
 }
 
+/// The timed comments of a catalogue track (`?id=`), oldest moment first.
+pub async fn comments(State(app): St, Extension(current): Extension<CurrentUser>, Query(q): Query<HashMap<String, String>>) -> Res {
+    let uid = match reader(&app, &current) {
+        Ok(u) => u.id,
+        Err(r) => return Ok(r),
+    };
+    let Some(id) = q.get("id").and_then(|v| v.parse::<i64>().ok()) else {
+        return Ok(StatusCode::BAD_REQUEST.into_response());
+    };
+    // Kept a minute in the browser too: scrolling back to a track asks again.
+    let ok = |c: &[crate::comments::Comment]| {
+        ([(axum::http::header::CACHE_CONTROL, "private, max-age=60")], Json(json!({ "comments": c }))).into_response()
+    };
+    if let Some(c) = app.comments.cached(id) {
+        return Ok(ok(&c));
+    }
+    // Only tracks in the catalogue: anything else would have us ask pl8list for nothing.
+    let Some((handle, slug)) = app.db.read(move |c| db::track_path(c, id)).await? else {
+        return Ok(StatusCode::NOT_FOUND.into_response());
+    };
+    if !app.limits.comments.allow(&uid.to_string()) {
+        return Ok(json_error(StatusCode::TOO_MANY_REQUESTS, "slow down a little"));
+    }
+    match app.comments.fetch(&app.pl8, id, &handle, &slug).await {
+        Ok(c) => Ok(ok(&c)),
+        Err(e) => {
+            tracing::warn!("comments of {handle}/{slug}: {e:#}");
+            Ok(StatusCode::BAD_GATEWAY.into_response())
+        }
+    }
+}
+
 /// How many neighbours the radio sheet shows, and how few count as still exploring.
 const SHEET_NEIGHBOURS: usize = 12;
 const FEW_NEIGHBOURS: usize = 3;

backend/src/http/mod.rs

@@ -28,6 +28,7 @@ use secrecy::SecretString;
 
 use crate::{
     cache::Cache,
+    comments::Comments,
     config::Config,
     crawler::Crawler,
     db::{self, Db, User},
@@ -61,6 +62,8 @@ pub struct Limits {
     pub resync: RateLimiter,
     /// Guests made per IP: each is a row in the database.
     pub guests: RateLimiter,
+    /// Comment fetches from pl8list, per user: only misses of the comment cache count.
+    pub comments: RateLimiter,
 }
 
 pub struct App {
@@ -75,6 +78,7 @@ pub struct App {
     pub cookies: Cookies,
     pub segments: proxy::SegmentKeys,
     pub legal: legal::Pages,
+    pub comments: Comments,
 }
 
 pub type St = State<Arc<App>>;
@@ -97,7 +101,9 @@ impl App {
                 pl8_writes: RateLimiter::new(60, 600),
                 resync: RateLimiter::new(1, 120),
                 guests: RateLimiter::new(10, 3600),
+                comments: RateLimiter::new(30, 300),
             },
+            comments: Comments::default(),
             logins: auth::PendingLogins::default(),
             segments: proxy::SegmentKeys::default(),
             sync,
@@ -119,7 +125,7 @@ impl App {
         }
         self.logins.prune();
         self.segments.prune();
-        for l in [&self.limits.guests, &self.limits.reads, &self.limits.proxy, &self.limits.proxy_ip, &self.limits.events] {
+        for l in [&self.limits.guests, &self.limits.reads, &self.limits.proxy, &self.limits.proxy_ip, &self.limits.events, &self.limits.comments] {
             l.prune();
         }
     }
@@ -349,6 +355,7 @@ pub fn router(app: Arc<App>) -> Router {
         .route("/local/likes", get(app::likes))
         .route("/local/track", get(app::track))
         .route("/local/artist", get(app::artist))
+        .route("/local/comments", get(app::comments))
         .route("/local/event", post(app::event))
         .route("/local/follow", post(app::follow))
         .route("/account/export", get(account::export))

backend/src/http/tests.rs

@@ -619,3 +619,33 @@ async fn radio_of_an_unknown_artist_is_not_found() {
     let res = r.clone().oneshot(req("GET", "/local/artist?handle=made-up-handle", None, Some(&guest), "203.0.113.23")).await.unwrap();
     assert_eq!(res.status(), StatusCode::NOT_FOUND);
 }
+
+#[tokio::test]
+async fn comments_come_sorted_by_moment_and_cached() {
+    let pl8 = MockServer::start().await;
+    let page = |n: i64, at: f64| {
+        serde_json::json!({ "totalPages": 2, "comments": [
+            { "id": n, "body": format!("c{n}"), "timestampSeconds": at, "user": { "handle": "h", "name": "H" } }
+        ]})
+    };
+    for (n, at) in [(1, 30.0), (2, 10.0)] {
+        Mock::given(method("GET")).and(path("/api/a/track-1/comments")).and(query_param("page", n.to_string()))
+            .respond_with(ResponseTemplate::new(200).set_body_json(page(n, at)))
+            .expect(1)
+            .mount(&pl8).await;
+    }
+    let (r, app, _) = setup("comments", &pl8.uri()).await;
+    known_track(&app, 1, "a").await;
+    let token = app.db.call(|c| { crate::db::upsert_user(c, 5, "tester", None, None)?; crate::db::create_session(c, 5) }).await.unwrap();
+    let cookie = format!("__Host-pl8inf={token}");
+    let get = |uri: &str| r.clone().oneshot(req("GET", uri, None, Some(&cookie), "203.0.113.1"));
+    for _ in 0..2 {
+        let res = get("/local/comments?id=1").await.unwrap();
+        assert_eq!(res.status(), StatusCode::OK);
+        let body = json_body(res).await;
+        let ids: Vec<i64> = body["comments"].as_array().unwrap().iter().map(|c| c["id"].as_i64().unwrap()).collect();
+        assert_eq!(ids, [2, 1]);
+    }
+    assert_eq!(get("/local/comments?id=99").await.unwrap().status(), StatusCode::NOT_FOUND);
+    assert_eq!(get("/local/comments?id=x").await.unwrap().status(), StatusCode::BAD_REQUEST);
+}

backend/src/main.rs

@@ -14,6 +14,7 @@
 #![allow(clippy::result_large_err)] // handlers return axum `Response` as their error
 
 mod cache;
+mod comments;
 mod config;
 mod crawler;
 mod db;

backend/src/pl8/mod.rs

@@ -156,6 +156,11 @@ impl Pl8 {
         self.get_json(&[handle, slug, "related"], &[], Some("routes/$handle_.$slug.related"), None).await
     }
 
+    /// One page of a track's comments, newest first (`comments`, `totalPages`).
+    pub async fn comments(&self, handle: &str, slug: &str, page: i64) -> anyhow::Result<Option<Value>> {
+        self.get_json(&["api", handle, slug, "comments"], &[("page", page.to_string())], None, None).await
+    }
+
     pub async fn best_selling_artists(&self) -> anyhow::Result<Option<Value>> {
         self.get_json(&["best-selling-artists"], &[], Some("routes/best-selling-artists"), None).await
     }