diff --git a/.gitignore b/.gitignore index f92620e..3970765 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,5 @@ __pycache__/ backend/target/ web/node_modules/ web/dist/ +backend/cache/ +cache/ diff --git a/README.md b/README.md index 781874d..e34d1ce 100644 --- a/README.md +++ b/README.md @@ -40,8 +40,8 @@ elsewhere). Debug builds of the server read `web/dist` from disk, so `npm run bu to see changes on :8790 too. The server is the Rust crate in `backend/`, the front end is TypeScript in `web/` (Vite, no -framework). The old Python server (`server.py` and friends) still works on the same database -but not with this front end; to roll back, check out `54a436a`. +framework). The old Python server is gone from the tree; it is in the history at `54a436a`, +and still opens the same database. Before going public in Germany, set the Impressum variables below. `/impressum` and `/datenschutz` are filled in from them at startup; the texts are in `backend/templates/`. diff --git a/backend/src/crawler.rs b/backend/src/crawler.rs index f3ded11..20ab1f6 100644 --- a/backend/src/crawler.rs +++ b/backend/src/crawler.rs @@ -52,6 +52,11 @@ const REFRESH_AFTER: f64 = 86400.0; // re-check an artist's newest page for uplo const RESEED_EVERY: Duration = Duration::from_secs(600); /// An artist radio crawls its artist's neighbourhood at most this often. const EXPLORE_EVERY: Duration = Duration::from_secs(600); +/// Jobs waiting at most. Past this only seed and feedback jobs get in, so nothing +/// from outside can grow the queue without end. +const MAX_QUEUE: usize = 20_000; +/// This many failed jobs in a row means pl8list is down or has changed its API. +const FAILURES_ALARM: u32 = 25; #[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug)] enum Job { @@ -95,6 +100,9 @@ impl Crawler { fn push(&self, priority: i64, job: Job) { let mut q = self.queue.lock().unwrap_or_else(|e| e.into_inner()); + if q.heap.len() >= MAX_QUEUE && priority > P_SEED { + return; + } if !q.queued.insert(job.clone()) { return; } @@ -153,7 +161,12 @@ impl Crawler { /// An artist radio was opened: crawl the artist, and the related pages of their /// best tracks that were not read yet, to fill in the artist graph around them. + /// Only artists already in the catalogue: a handle typed into a URL queues nothing. pub async fn explore_artist(&self, handle: String) { + let h = handle.clone(); + if !self.db.read(move |c| db::artist_known(c, &h)).await.unwrap_or(false) { + return; + } { let mut explored = self.explored.lock().unwrap_or_else(|e| e.into_inner()); let now = Instant::now(); @@ -187,10 +200,26 @@ impl Crawler { async fn run(&self) { self.seed(); let mut last_seed = Instant::now(); + let mut failures = 0u32; loop { let job = self.pop().await; - if let Err(e) = self.run_job(job.clone()).await { - tracing::warn!("crawl {job:?} failed: {e:#}"); + match self.run_job(job.clone()).await { + Ok(()) => { + if failures >= FAILURES_ALARM { + tracing::info!("crawler: pl8list answers again after {failures} failed jobs"); + } + failures = 0; + } + Err(e) => { + tracing::warn!("crawl {job:?} failed: {e:#}"); + failures += 1; + if failures == FAILURES_ALARM { + tracing::error!( + "crawler: {failures} jobs in a row failed. pl8list is down, or its routes changed \ + (a move from Remix `?_data=` to React Router `.data` would look like this)" + ); + } + } } if !self.busy().await { tokio::time::sleep(IDLE_DELAY).await; @@ -308,10 +337,18 @@ impl Crawler { match job { Job::Home => { let d = self.pl8.home().await?.unwrap_or(Value::Null); + let mut found = 0; for section in ["bestSellers", "recommended"] { - self.upsert(normalize::items(d.get(section).and_then(|s| s.get("items"))), None).await?; + let rows = normalize::items(d.get(section).and_then(|s| s.get("items"))); + found += rows.len(); + self.upsert(rows, None).await?; } - self.ticker_events(d.get("tickerEvents")).await + self.ticker_events(d.get("tickerEvents")).await?; + // The home page always lists tracks; none means its shape changed. + if found == 0 { + anyhow::bail!("home page has no tracks in bestSellers or recommended; did pl8list change its data?"); + } + Ok(()) } Job::Ticker => { let d = self.pl8.activity_ticker().await?.unwrap_or(Value::Null); diff --git a/backend/src/db.rs b/backend/src/db.rs index 3185263..6b50dbe 100644 --- a/backend/src/db.rs +++ b/backend/src/db.rs @@ -5,13 +5,16 @@ use std::{ path::Path, - sync::{Arc, Mutex}, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, time::{SystemTime, UNIX_EPOCH}, }; use base64::Engine; use rand::Rng; -use rusqlite::{Connection, OptionalExtension, Row, params, params_from_iter}; +use rusqlite::{Connection, OpenFlags, OptionalExtension, Row, params, params_from_iter}; use serde::Serialize; use sha2::{Digest, Sha256}; @@ -134,8 +137,17 @@ pub fn now() -> f64 { SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64() } +/// Read-only connections next to the one that writes. With WAL they read while it +/// writes, so a slow feed query does not hold up every other request. +const READERS: usize = 4; + +/// One connection that writes, and a few that only read. #[derive(Clone)] -pub struct Db(Arc>); +pub struct Db { + writer: Arc>, + readers: Arc>>, + next: Arc, +} impl Db { pub fn open(path: &Path) -> anyhow::Result { @@ -148,16 +160,22 @@ impl Db { if tagged > 0 { tracing::info!("tagged {tagged} tracks"); } - Ok(Db(Arc::new(Mutex::new(conn)))) + let mut readers = Vec::with_capacity(READERS); + for _ in 0..READERS { + let r = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX)?; + r.busy_timeout(std::time::Duration::from_secs(5))?; + readers.push(Mutex::new(r)); + } + Ok(Db { writer: Arc::new(Mutex::new(conn)), readers: Arc::new(readers), next: Arc::new(AtomicUsize::new(0)) }) } - /// Run `f` on the connection off the async runtime. + /// Run `f` on the writing connection off the async runtime. pub async fn call(&self, f: F) -> anyhow::Result where T: Send + 'static, F: FnOnce(&mut Connection) -> rusqlite::Result + Send + 'static, { - let db = self.0.clone(); + let db = self.writer.clone(); let out = tokio::task::spawn_blocking(move || { let mut conn = db.lock().unwrap_or_else(|e| e.into_inner()); f(&mut conn) @@ -165,6 +183,25 @@ impl Db { .await??; Ok(out) } + + /// Run a read-only `f` on one of the reading connections: a free one if there + /// is one, else the next in turn. + pub async fn read(&self, f: F) -> anyhow::Result + where + T: Send + 'static, + F: FnOnce(&Connection) -> rusqlite::Result + Send + 'static, + { + let readers = self.readers.clone(); + let start = self.next.fetch_add(1, Ordering::Relaxed); + let out = tokio::task::spawn_blocking(move || { + let n = readers.len(); + let free = (0..n).find_map(|i| readers[(start + i) % n].try_lock().ok()); + let conn = free.unwrap_or_else(|| readers[start % n].lock().unwrap_or_else(|e| e.into_inner())); + f(&conn) + }) + .await??; + Ok(out) + } } fn migrate(conn: &Connection) -> rusqlite::Result<()> { @@ -298,6 +335,14 @@ pub fn artist(conn: &Connection, handle: &str) -> rusqlite::Result rusqlite::Result { + conn.prepare_cached( + "SELECT 1 WHERE EXISTS (SELECT 1 FROM tracks WHERE handle=?1) OR EXISTS (SELECT 1 FROM artists WHERE handle=?1)", + )? + .exists([handle]) +} + /// An artist's most common style tags; tempo tags left out. pub fn artist_tags(conn: &Connection, handle: &str, n: usize) -> rusqlite::Result> { conn.prepare_cached( @@ -564,6 +609,25 @@ pub fn create_guest(conn: &Connection) -> rusqlite::Result<(User, String)> { Ok((user, token)) } +/// What a guest would add to a pl8list library: their liked tracks (`[handle, slug]`, +/// newest first, not unliked since) and followed artists (`[handle]`), at most `max`. +pub fn guest_library(conn: &Connection, guest_id: i64, max: usize) -> rusqlite::Result>> { + let liked: Vec> = conn + .prepare( + "SELECT t.handle, t.slug FROM events e JOIN tracks t ON t.id = e.track_id \ + WHERE e.user_id=?1 AND e.kind='like' AND NOT EXISTS (SELECT 1 FROM events u \ + WHERE u.user_id=?1 AND u.track_id=e.track_id AND u.kind='unlike' AND u.ts >= e.ts) \ + GROUP BY e.track_id ORDER BY MAX(e.ts) DESC", + )? + .query_map([guest_id], |r| Ok(vec![r.get(0)?, r.get(1)?]))? + .collect::>()?; + let followed: Vec> = conn + .prepare("SELECT handle FROM follows WHERE user_id=? ORDER BY ts DESC")? + .query_map([guest_id], |r| Ok(vec![r.get(0)?]))? + .collect::>()?; + Ok(followed.into_iter().chain(liked).take(max).collect()) +} + /// A guest signed in with pl8list: what they played, liked and followed moves to /// the account, and the guest is gone. pub fn merge_guest(conn: &mut Connection, guest_id: i64, user_id: i64) -> rusqlite::Result<()> { @@ -601,6 +665,7 @@ pub fn create_session(conn: &Connection, user_id: i64) -> rusqlite::Result, } +/// The session behind a token, read-only. `touch_session` then records the use. pub fn session_user(conn: &Connection, token: &str) -> rusqlite::Result> { let now = now(); - let hit = conn + conn .query_row( "SELECT u.*, s.created_at AS s_created, s.expires_at AS s_expires, s.pl8_link, s.pl8_synced_at \ FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token_hash=? AND s.expires_at > ?", @@ -626,31 +692,46 @@ pub fn session_user(conn: &Connection, token: &str) -> rusqlite::Result ?", - params![now + SESSION_IDLE_DAYS * 86400.0, hash_token(token), now + 120.0], - )?; - } + .optional() +} + +impl SessionHit { + /// Whether `touch_session` has anything to write: last seen at most once a + /// minute, the sliding expiry at most once a day. + pub fn stale(&self) -> bool { + let now = now(); + self.user.last_seen.unwrap_or(0.0) < now - 60.0 || self.expires_at < now + (SESSION_IDLE_DAYS - 1.0) * 86400.0 } - Ok(hit) +} + +pub fn touch_session(conn: &Connection, token: &str, hit: &SessionHit) -> rusqlite::Result<()> { + let now = now(); + if hit.user.last_seen.unwrap_or(0.0) < now - 60.0 { + conn.execute("UPDATE users SET last_seen=? WHERE id=?", params![now, hit.user.id])?; + } + if hit.expires_at < now + (SESSION_IDLE_DAYS - 1.0) * 86400.0 { + // Not a session that was just rotated away: that one stays short. + conn.execute( + "UPDATE sessions SET expires_at=? WHERE token_hash=? AND expires_at > ?", + params![now + SESSION_IDLE_DAYS * 86400.0, hash_token(token), now + 120.0], + )?; + } + Ok(()) } /// Replace a session with a new token. The old one keeps working for a minute so -/// requests already in flight do not fail. -pub fn rotate_session(conn: &Connection, old_token: &str, user_id: i64) -> rusqlite::Result { - let token = create_session(conn, user_id)?; - conn.execute( - "UPDATE sessions SET expires_at=MIN(expires_at, ?) WHERE token_hash=?", - params![now() + 60.0, hash_token(old_token)], +/// requests already in flight do not fail. Returns `None` when another request +/// rotated it already: then that request's response carries the new token. +pub fn rotate_session(conn: &Connection, old_token: &str, user_id: i64) -> rusqlite::Result> { + let now = now(); + let retired = conn.execute( + "UPDATE sessions SET expires_at=? WHERE token_hash=? AND user_id=? AND expires_at > ?", + params![now + 60.0, hash_token(old_token), user_id, now + 60.0], )?; - Ok(token) + if retired == 0 { + return Ok(None); + } + create_session(conn, user_id).map(Some) } /// Store (or replace) the encrypted pl8list session of one device. @@ -856,8 +937,11 @@ mod tests { assert_eq!(stored, hash_token(&token)); assert_eq!(session_user(&c, &token).unwrap().unwrap().user.id, 7); - let new = rotate_session(&c, &token, 7).unwrap(); + let new = rotate_session(&c, &token, 7).unwrap().unwrap(); assert!(session_user(&c, &new).unwrap().is_some()); + // A second request with the old token, running alongside, does not rotate again. + assert_eq!(rotate_session(&c, &token, 7).unwrap(), None); + assert_eq!(c.query_row("SELECT COUNT(*) FROM sessions", [], |r| r.get::<_, i64>(0)).unwrap(), 2); let old_expiry: f64 = c .query_row("SELECT expires_at FROM sessions WHERE token_hash=?", [hash_token(&token)], |r| r.get(0)) .unwrap(); diff --git a/backend/src/http/app.rs b/backend/src/http/app.rs index b3ad532..2e77726 100644 --- a/backend/src/http/app.rs +++ b/backend/src/http/app.rs @@ -52,13 +52,22 @@ async fn resync_if_due(app: &App, current: &CurrentUser) { } } +/// The signed-in user, within the limit for catalogue reads. +fn reader(app: &App, current: &CurrentUser) -> Result { + let user = current.require()?; + if !app.limits.reads.allow(&user.id.to_string()) { + return Err(json_error(StatusCode::TOO_MANY_REQUESTS, "slow down a little")); + } + Ok(user.clone()) +} + fn num(q: &HashMap, name: &str) -> Option { q.get(name).and_then(|v| v.parse::().ok()).filter(|v| v.is_finite()) } pub async fn feed(State(app): St, Extension(current): Extension, Query(q): Query>) -> Res { - let user = match current.require() { - Ok(u) => u.clone(), + let user = match reader(&app, ¤t) { + Ok(u) => u, Err(r) => return Ok(r), }; let artist = q.get("artist").map(String::as_str).filter(|h| valid_handle(h)); @@ -78,11 +87,18 @@ pub async fn feed(State(app): St, Extension(current): Extension, Qu resync_if_due(&app, ¤t).await; let deadline = Instant::now() + FEED_WAIT; let uid = user.id; - let mut tracks = loop { + let (mut tracks, liked) = loop { let mode = mode.clone(); - let tracks = app.db.call(move |c| ranker::pick(c, uid, n, filters, &*mode.ranker(c)?)).await?; + let (tracks, liked) = app + .db + .read(move |c| { + let taste = ranker::taste(c, uid)?; + let tracks = ranker::pick_with(c, uid, n, filters, &*mode.ranker(c)?, &taste)?; + Ok((tracks, taste.liked_ids.len())) + }) + .await?; if !tracks.is_empty() || Instant::now() >= deadline { - break tracks; + break (tracks, liked); } tokio::time::sleep(Duration::from_millis(1500)).await; }; @@ -95,7 +111,7 @@ pub async fn feed(State(app): St, Extension(current): Extension, Qu db::add_event(c, uid, id, "served", None)?; } let mut s = db::stats(c, uid)?; - s.liked = ranker::taste(c, uid)?.liked_ids.len(); + s.liked = liked; Ok(s) }) .await?; @@ -103,11 +119,11 @@ pub async fn feed(State(app): St, Extension(current): Extension, Qu } pub async fn likes(State(app): St, Extension(current): Extension) -> Res { - let uid = match current.require() { + let uid = match reader(&app, ¤t) { Ok(u) => u.id, Err(r) => return Ok(r), }; - let mut tracks = app.db.call(move |c| ranker::likes(c, uid)).await?; + let mut tracks = app.db.read(move |c| ranker::likes(c, uid)).await?; mark_followed(&app, uid, &mut tracks).await?; Ok((no_store(), Json(json!({ "tracks": tracks }))).into_response()) } @@ -129,24 +145,27 @@ pub async fn event(State(app): St, Extension(current): Extension, J if !app.limits.events.allow(&uid.to_string()) { return Ok(StatusCode::TOO_MANY_REQUESTS.into_response()); } + // Only tracks in the catalogue: anything else would be a row nothing reads. + let Some((handle, slug)) = app.db.read(move |c| db::track_path(c, track_id)).await? else { + return Ok(StatusCode::NOT_FOUND.into_response()); + }; let kind = kind.to_string(); let k = kind.clone(); app.db.call(move |c| db::add_event(c, uid, track_id, &k, seconds)).await?; if kind == "like" || kind == "complete" { app.crawler.on_positive(track_id).await; } - if kind == "like" && app.limits.pl8_writes.allow(&uid.to_string()) { - let row = app.db.call(move |c| db::track_path(c, track_id)).await?; - if let (Some((handle, slug)), Some((sealer, session))) = (row, app.connected(¤t).await) { - app.sync.push_like(sealer, session, handle, slug); - } + if kind == "like" && app.limits.pl8_writes.allow(&uid.to_string()) + && let Some((sealer, session)) = app.connected(¤t).await + { + app.sync.push_like(sealer, session, handle, slug); } Ok((no_store(), Json(json!({ "ok": true }))).into_response()) } /// One track, for a link into the app (`/?t=handle/slug`). pub async fn track(State(app): St, Extension(current): Extension, Query(q): Query>) -> Res { - let uid = match current.require() { + let uid = match reader(&app, ¤t) { Ok(u) => u.id, Err(r) => return Ok(r), }; @@ -155,7 +174,7 @@ pub async fn track(State(app): St, Extension(current): Extension, Q let (Some(handle), Some(slug)) = (handle, slug) else { return Ok(StatusCode::BAD_REQUEST.into_response()); }; - let Some(track) = app.db.call(move |c| ranker::track_by_path(c, &handle, &slug)).await? else { + let Some(track) = app.db.read(move |c| ranker::track_by_path(c, &handle, &slug)).await? else { return Ok(json_error(StatusCode::NOT_FOUND, "no such track here (yet)")); }; let mut tracks = [track]; @@ -172,7 +191,7 @@ const FEW_NEIGHBOURS: usize = 3; /// Starts crawling the neighbourhood when it is new, so the sheet can poll while /// `exploring` is true. pub async fn artist(State(app): St, Extension(current): Extension, Query(q): Query>) -> Res { - let uid = match current.require() { + let uid = match reader(&app, ¤t) { Ok(u) => u.id, Err(r) => return Ok(r), }; @@ -182,7 +201,7 @@ pub async fn artist(State(app): St, Extension(current): Extension, app.crawler.explore_artist(handle.clone()).await; let body = app .db - .call(move |c| { + .read(move |c| { let Some(artist) = db::artist(c, &handle)? else { return Ok(None) }; let mut neighbours = vec![]; for (h, weight) in ranker::scaled_neighbours(c, &handle)?.into_iter().take(SHEET_NEIGHBOURS) { @@ -226,6 +245,10 @@ pub async fn follow(State(app): St, Extension(current): Extension, return Ok(StatusCode::TOO_MANY_REQUESTS.into_response()); } let h = handle.clone(); + if on && !app.db.read(move |c| db::artist_known(c, &h)).await? { + return Ok(json_error(StatusCode::NOT_FOUND, "unknown artist")); + } + let h = handle.clone(); let added = app .db .call(move |c| if on { db::add_follows(c, uid, &[h], "local") } else { db::unfollow(c, uid, &h).map(|_| 0) }) diff --git a/backend/src/http/auth.rs b/backend/src/http/auth.rs index 8b9ff10..52db2b0 100644 --- a/backend/src/http/auth.rs +++ b/backend/src/http/auth.rs @@ -18,7 +18,7 @@ use axum::{ use serde_json::{Value, json}; use super::{App, CurrentUser, LOGIN_COOKIE, Peer, Res, SESSION_COOKIE, St, client_ip, json_error, no_store, security}; -use crate::{db, pl8::login::LoginFlow}; +use crate::{db, pl8::login::LoginFlow, sync}; const LOGIN_TTL: Duration = Duration::from_secs(900); const MAX_PENDING: usize = 500; @@ -164,17 +164,22 @@ pub async fn verify(State(app): St, Extension(current): Extension, } // A guest signing in keeps what they played, liked and followed. let guest = current.user.as_ref().filter(|g| g.is_guest()).map(|g| g.id); - let token = app + let (token, guest_library) = app .db .call(move |c| { if let Some(t) = previous { db::delete_session(c, &t)?; } db::upsert_user(c, u.id, &u.handle, u.name.as_deref(), u.img.as_deref())?; + // Staying connected, the guest's likes and follows go to pl8list too. + let mut library = vec![]; if let Some(g) = guest { + if stay { + library = db::guest_library(c, g, sync::MAX_GUEST_PUSH)?; + } db::merge_guest(c, g, u.id)?; } - db::create_session(c, u.id) + Ok((db::create_session(c, u.id)?, library)) }) .await?; if stay { @@ -182,7 +187,7 @@ pub async fn verify(State(app): St, Extension(current): Extension, let blob = sealer.seal(&pl8_session); let hash = sealer.token_hash.clone(); app.db.call(move |c| db::set_link(c, &hash, &blob, Some(db::now()))).await?; - app.sync.linked(sealer, pl8_session); + app.sync.linked_with(sealer, pl8_session, guest_library); } else { app.sync.once(user.id, pl8_session); } diff --git a/backend/src/http/mod.rs b/backend/src/http/mod.rs index 8fb2f99..f41a5e1 100644 --- a/backend/src/http/mod.rs +++ b/backend/src/http/mod.rs @@ -49,7 +49,12 @@ pub struct Limits { /// Every start and resend makes pl8list send a mail; cap them for the whole server. pub login_global: RateLimiter, pub events: RateLimiter, + /// Feed, artist, track and likes: each runs queries over the catalogue. + pub reads: RateLimiter, pub proxy: RateLimiter, + /// The proxy per address too: guests are cheap to make, so a per-user limit + /// alone would not hold. + pub proxy_ip: RateLimiter, /// Saves to a user's pl8list library (likes and follows pushed from here). pub pl8_writes: RateLimiter, /// The resync button. @@ -86,7 +91,9 @@ impl App { login_email: RateLimiter::new(3, 900), login_global: RateLimiter::new(30, 600), events: RateLimiter::new(600, 600), + reads: RateLimiter::new(120, 60), proxy: RateLimiter::new(1200, 60), + proxy_ip: RateLimiter::new(3000, 60), pl8_writes: RateLimiter::new(60, 600), resync: RateLimiter::new(1, 120), guests: RateLimiter::new(10, 3600), @@ -112,7 +119,9 @@ impl App { } self.logins.prune(); self.segments.prune(); - self.limits.guests.prune(); + for l in [&self.limits.guests, &self.limits.reads, &self.limits.proxy, &self.limits.proxy_ip, &self.limits.events] { + l.prune(); + } } /// This device's stored pl8list session, if it stays connected. A blob that no @@ -190,7 +199,13 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response { let mut current = CurrentUser { user: None, token: token.clone(), link: None, synced_at: None }; if let Some(t) = token { let lookup = t.clone(); - if let Ok(Some(hit)) = app.db.call(move |c| db::session_user(c, &lookup)).await { + // Looked up on a reader; the writer only sees the rare request that has to + // record something. + if let Ok(Some(hit)) = app.db.read(move |c| db::session_user(c, &lookup)).await { + if hit.stale() { + let (tt, h) = (t.clone(), hit.clone()); + let _ = app.db.call(move |c| db::touch_session(c, &tt, &h)).await; + } let mut new_token = None; current.link = hit.link.clone(); current.synced_at = hit.synced_at; @@ -201,7 +216,7 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response { let key = &app.sync.key; let synced = hit.synced_at; new_token = match app.db.call(move |c| db::rotate_session(c, &old, uid)).await { - Ok(nt) => { + Ok(Some(nt)) => { if let Some(session) = reopened { let new = key.sealer(&nt, uid); let blob = new.seal(&session); @@ -220,7 +235,9 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response { } Some(nt) } - Err(_) => None, + // Rotated by a request running alongside this one, or failed: the + // old token works for another minute either way. + Ok(None) | Err(_) => None, }; } else if legacy { new_token = Some(t.clone()); @@ -311,14 +328,17 @@ async fn healthz() -> Response { pub fn router(app: Arc) -> Router { let secure = app.cfg.secure; - Router::new() + // The page itself, its assets and the legal pages need no session: no database + // lookup for each of them. + let public = Router::new() .route("/", get(index)) .route("/static/{*path}", get(static_route)) .route("/sw.js", get(service_worker)) .route("/healthz", get(healthz)) .route("/impressum", get(legal::impressum)) .route("/datenschutz", get(legal::datenschutz)) - .route("/privacy", get(legal::datenschutz)) + .route("/privacy", get(legal::datenschutz)); + let api = Router::new() .route("/auth/start", post(auth::start)) .route("/auth/resend", post(auth::resend)) .route("/auth/verify", post(auth::verify)) @@ -339,7 +359,9 @@ pub fn router(app: Arc) -> Router { .route("/api/{handle}/{slug}/hls/playlist.m3u8", get(proxy::playlist)) .route("/api/{handle}/{slug}/hls/segment/{n}", get(proxy::segment)) .route("/img/{id}", get(proxy::image)) - .layer(middleware::from_fn_with_state(app.clone(), session)) + .layer(middleware::from_fn_with_state(app.clone(), session)); + public + .merge(api) .layer(middleware::from_fn_with_state(app.clone(), origin_check)) .layer(DefaultBodyLimit::max(4096)) .layer(TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, Duration::from_secs(30))) diff --git a/backend/src/http/proxy.rs b/backend/src/http/proxy.rs index 56d0f4c..1df0850 100644 --- a/backend/src/http/proxy.rs +++ b/backend/src/http/proxy.rs @@ -12,6 +12,7 @@ use std::{ collections::HashMap, + net::SocketAddr, sync::Mutex, time::{Duration, Instant}, }; @@ -19,7 +20,7 @@ use std::{ use axum::{ Extension, extract::{Path, Query, State}, - http::{StatusCode, header}, + http::{HeaderMap, StatusCode, header}, response::{IntoResponse, Response}, }; use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; @@ -28,7 +29,7 @@ use futures_util::StreamExt; use serde::Deserialize; use serde_json::Value; -use super::{App, CurrentUser, Res, St}; +use super::{App, CurrentUser, Peer, Res, St, client_ip}; use crate::cache::{self, IMAGE_TYPES, SEGMENT_TYPES}; const MAX_PLAYLIST: usize = 256 << 10; @@ -82,9 +83,9 @@ fn token_upload(token: &str) -> Option { v.get("u")?.as_str().filter(|u| safe_segment(u, 128)).map(String::from) } -fn require_user(app: &App, current: &CurrentUser) -> Result<(), Response> { +fn require_user(app: &App, current: &CurrentUser, headers: &HeaderMap, peer: SocketAddr) -> Result<(), Response> { let user = current.require()?; - if !app.limits.proxy.allow(&user.id.to_string()) { + if !app.limits.proxy.allow(&user.id.to_string()) || !app.limits.proxy_ip.allow(&client_ip(app, headers, peer)) { return Err(StatusCode::TOO_MANY_REQUESTS.into_response()); } Ok(()) @@ -133,10 +134,12 @@ pub struct PlaylistQuery { pub async fn playlist( State(app): St, Extension(current): Extension, + peer: Peer, + headers: HeaderMap, Path((handle, slug)): Path<(String, String)>, Query(q): Query, ) -> Res { - if let Err(r) = require_user(&app, ¤t) { + if let Err(r) = require_user(&app, ¤t, &headers, peer.0) { return Ok(r); } if !safe_segment(&handle, 64) || !safe_segment(&slug, 160) || q.v.as_deref().is_some_and(|v| !safe_segment(v, 128)) { @@ -180,10 +183,12 @@ pub struct SegmentQuery { pub async fn segment( State(app): St, Extension(current): Extension, + peer: Peer, + headers: HeaderMap, Path((handle, slug, n)): Path<(String, String, String)>, Query(q): Query, ) -> Res { - if let Err(r) = require_user(&app, ¤t) { + if let Err(r) = require_user(&app, ¤t, &headers, peer.0) { return Ok(r); } let token = q.st.unwrap_or_default(); @@ -219,8 +224,8 @@ pub async fn segment( Ok(media(ctype, body, "private, max-age=86400")) } -pub async fn image(State(app): St, Extension(current): Extension, Path(id): Path) -> Res { - if let Err(r) = require_user(&app, ¤t) { +pub async fn image(State(app): St, Extension(current): Extension, peer: Peer, headers: HeaderMap, Path(id): Path) -> Res { + if let Err(r) = require_user(&app, ¤t, &headers, peer.0) { return Ok(r); } if id.len() > 64 || id.is_empty() || !id.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') { diff --git a/backend/src/http/tests.rs b/backend/src/http/tests.rs index d997027..08d2fa6 100644 --- a/backend/src/http/tests.rs +++ b/backend/src/http/tests.rs @@ -301,6 +301,7 @@ async fn connected_device_syncs_library_and_pushes_likes_and_follows() { let res = r.clone().oneshot(req("POST", "/local/event", Some(r#"{"track_id":77,"kind":"like"}"#), Some(&session), "203.0.113.9")).await.unwrap(); assert_eq!(res.status(), StatusCode::OK); + known_track(&app, 78, "artist3").await; let res = r.clone().oneshot(req("POST", "/local/follow", Some(r#"{"handle":"artist3","follow":true}"#), Some(&session), "203.0.113.9")).await.unwrap(); assert_eq!(res.status(), StatusCode::OK); assert!(wait_for(&pl8, |q| q.url.path() == "/api/artist1/song/add-to-library").await); @@ -507,10 +508,17 @@ async fn artist_sheet_and_radio_reasons() { assert_eq!(why[&3], serde_json::json!({ "kind": "via", "handle": "b", "name": "B" })); } +/// A track (and so its artist) in the catalogue; events and follows only take known ones. +async fn known_track(app: &crate::http::App, id: i64, handle: &str) { + let t = crate::db::Track { id, handle: handle.into(), artist_name: handle.into(), slug: format!("track-{id}"), title: "T".into(), ..Default::default() }; + app.db.call(move |c| crate::db::upsert_tracks(c, &[t], None)).await.unwrap(); +} + #[tokio::test] async fn guests_listen_and_keep_it_all_when_they_sign_in() { let pl8 = fake_pl8list(1).await; let (r, app, _) = setup("guest", &pl8.uri()).await; + known_track(&app, 7, "motus").await; let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.20")).await.unwrap(); assert_eq!(res.status(), StatusCode::OK); let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf"); @@ -521,6 +529,11 @@ async fn guests_listen_and_keep_it_all_when_they_sign_in() { let res = r.clone().oneshot(req("POST", uri, Some(body), Some(&guest), "203.0.113.20")).await.unwrap(); assert_eq!(res.status(), StatusCode::OK, "{uri}"); } + // Nothing the catalogue does not know. + for (uri, body) in [("/local/event", r#"{"track_id":999,"kind":"like"}"#), ("/local/follow", r#"{"handle":"nobody-here","follow":true}"#)] { + let res = r.clone().oneshot(req("POST", uri, Some(body), Some(&guest), "203.0.113.20")).await.unwrap(); + assert_eq!(res.status(), StatusCode::NOT_FOUND, "{uri}"); + } // Signing in on the same device takes the guest's likes and follows along. let res = r.clone().oneshot(req("POST", "/auth/start", Some(r#"{"email":"a@b.de"}"#), None, "203.0.113.20")).await.unwrap(); @@ -550,6 +563,7 @@ async fn guests_listen_and_keep_it_all_when_they_sign_in() { #[tokio::test] async fn guests_leave_nothing_behind_and_are_capped_per_ip() { let (r, app, _) = setup("guest-gone", "http://127.0.0.1:9").await; + known_track(&app, 7, "motus").await; let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.21")).await.unwrap(); let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf"); // A device with a session gets no second guest. @@ -576,3 +590,32 @@ async fn guests_leave_nothing_behind_and_are_capped_per_ip() { } assert_eq!(statuses.iter().filter(|s| **s == StatusCode::TOO_MANY_REQUESTS).count(), 2); } + +#[tokio::test] +async fn guest_staying_connected_saves_what_they_liked_to_pl8list() { + let pl8 = fake_pl8list(0).await; + let (r, app, _) = setup("guest-push", &pl8.uri()).await; + known_track(&app, 7, "motus").await; + let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.22")).await.unwrap(); + let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf"); + for (uri, body) in [("/local/event", r#"{"track_id":7,"kind":"like"}"#), ("/local/follow", r#"{"handle":"motus","follow":true}"#)] { + let res = r.clone().oneshot(req("POST", uri, Some(body), Some(&guest), "203.0.113.22")).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK, "{uri}"); + } + let res = r.clone().oneshot(req("POST", "/auth/start", Some(r#"{"email":"a@b.de"}"#), None, "203.0.113.22")).await.unwrap(); + let login = cookie_pair(&set_cookies(&res), "__Host-pl8inf_login"); + let both = format!("{login}; {guest}"); + let res = r.clone().oneshot(req("POST", "/auth/verify", Some(r#"{"code":"123456","connect":true}"#), Some(&both), "203.0.113.22")).await.unwrap(); + assert_eq!(res.status(), StatusCode::OK); + assert!(wait_for(&pl8, |q| q.url.path() == "/api/motus/add-to-library").await, "follow not saved"); + assert!(wait_for(&pl8, |q| q.url.path() == "/api/motus/track-7/add-to-library").await, "like not saved"); +} + +#[tokio::test] +async fn radio_of_an_unknown_artist_is_not_found() { + let (r, _app, _) = setup("explore-unknown", "http://127.0.0.1:9").await; + let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.23")).await.unwrap(); + let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf"); + 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); +} diff --git a/backend/src/ranker.rs b/backend/src/ranker.rs index b41dec1..7b98352 100644 --- a/backend/src/ranker.rs +++ b/backend/src/ranker.rs @@ -349,7 +349,13 @@ pub struct Filters { pub bpm_max: Option, } +#[cfg(test)] pub fn pick(conn: &Connection, user_id: i64, n: usize, f: Filters, ranker: &dyn Ranker) -> rusqlite::Result> { + pick_with(conn, user_id, n, f, ranker, &taste(conn, user_id)?) +} + +/// `pick` with the user's taste worked out already. +pub fn pick_with(conn: &Connection, user_id: i64, n: usize, f: Filters, ranker: &dyn Ranker, t: &Taste) -> rusqlite::Result> { let mut sql = String::from( "SELECT * FROM tracks t WHERE (t.full_access=1 OR t.preview_upload_id IS NOT NULL) \ AND NOT EXISTS (SELECT 1 FROM events e WHERE e.user_id=? AND e.kind='served' AND e.track_id=t.id)", @@ -377,7 +383,6 @@ pub fn pick(conn: &Connection, user_id: i64, n: usize, f: Filters, ranker: &dyn return Ok(vec![]); } - let t = taste(conn, user_id)?; let recent: Vec = conn .prepare_cached( "SELECT t.handle FROM events e JOIN tracks t ON t.id=e.track_id \ diff --git a/backend/src/sync.rs b/backend/src/sync.rs index 4437607..828e28b 100644 --- a/backend/src/sync.rs +++ b/backend/src/sync.rs @@ -29,6 +29,8 @@ use crate::{ /// A connected device resyncs on use once its last sync is this old. pub const RESYNC_AFTER: f64 = 6.0 * 3600.0; const MAX_PAGES: i64 = 20; +/// A guest's likes and follows saved to pl8list when they sign in and stay connected. +pub const MAX_GUEST_PUSH: usize = 200; const DEADLINE: Duration = Duration::from_secs(120); pub struct Syncer { @@ -76,15 +78,21 @@ impl Syncer { /// Sync once with a fresh sign-in session, then log it out at pl8list. pub fn once(self: &Arc, user_id: i64, session: SecretString) { - self.spawn_sync(user_id, session, After::Logout); + self.spawn_sync(user_id, session, After::Logout, vec![]); } /// Sync with a connected device's session. Returns false if one is already running. pub fn linked(self: &Arc, sealer: Sealer, session: SecretString) -> bool { - self.spawn_sync(sealer.user_id, session, After::Store(sealer)) + self.spawn_sync(sealer.user_id, session, After::Store(sealer), vec![]) } - fn spawn_sync(self: &Arc, user_id: i64, session: SecretString, after: After) -> bool { + /// Like `linked`, saving `push` (paths like `[handle]` or `[handle, slug]`) to the + /// pl8list library first: what a guest liked and followed before signing in. + pub fn linked_with(self: &Arc, sealer: Sealer, session: SecretString, push: Vec>) -> bool { + self.spawn_sync(sealer.user_id, session, After::Store(sealer), push) + } + + fn spawn_sync(self: &Arc, user_id: i64, session: SecretString, after: After, push: Vec>) -> bool { if !self.claim(user_id) { if let After::Logout = after { // Never keep a one-time session around just because a sync is running. @@ -102,7 +110,11 @@ impl Syncer { } } let mut s = UserSession::new(session); - let res = tokio::time::timeout(DEADLINE, me.read_account(user_id, &mut s)).await; + let res = tokio::time::timeout(DEADLINE, async { + me.push_all(&mut s, &push).await?; + me.read_account(user_id, &mut s).await + }) + .await; let expired = match &res { Ok(Ok((feed, likes, follows))) => { tracing::info!("synced pl8list: {feed} feed pages, {likes} new likes, {follows} new follows"); @@ -180,6 +192,24 @@ impl Syncer { } } + /// Save each path to the library, one after the other. A failed save is skipped; + /// an expired session ends it. + async fn push_all(&self, s: &mut UserSession, push: &[Vec]) -> anyhow::Result<()> { + let mut saved = 0; + for path in push { + let parts: Vec<&str> = path.iter().map(String::as_str).collect(); + match self.pl8.save(s, &parts).await { + Ok(()) => saved += 1, + Err(e) if e.is::() => return Err(e), + Err(e) => tracing::debug!("pl8list save failed: {e:#}"), + } + } + if !push.is_empty() { + tracing::info!("saved {saved} of {} guest likes and follows to pl8list", push.len()); + } + Ok(()) + } + /// Feed, saved tracks and saved artists. Returns (feed pages, new likes, new follows). async fn read_account(&self, user_id: i64, s: &mut UserSession) -> anyhow::Result<(i64, usize, usize)> { let mut feed_pages = 0; diff --git a/crawler.py b/crawler.py deleted file mode 100644 index d0e9bb5..0000000 --- a/crawler.py +++ /dev/null @@ -1,251 +0,0 @@ -"""Background discovery crawler. - -There is no "all tracks" endpoint, so the pool grows by snowballing: -home page sections, the activity ticker, best-selling artists and user search -give us artists; artist pages give us their tracks; "related" pages of tracks -users liked or finished pull in neighbours. Everything goes through one priority -queue so feedback is crawled first. - -The catalogue is shared by all users and crawled as a guest. The crawler runs -at full pace only while someone has used the app recently. -""" - -import heapq -import itertools -import logging -import threading -import time - -from pl8list import Client, normalize_item, normalize_related - -log = logging.getLogger("crawler") - -# Search needs at least 3 characters. Genre-ish words find lots of artist handles. -SEARCH_TERMS = [ - "dub", "bass", "dnb", "jungle", "garage", "grime", "riddim", "dubstep", "trap", "techno", - "house", "beats", "music", "sound", "prod", "audio", "records", "drill", "breaks", "wave", - "deep", "dark", "the", "official", "dj", "mc", "lofi", "halftime", "neuro", "tek", -] - -# Lower number = sooner. -P_FEEDBACK = 0 -P_SEED = 10 -P_ARTIST_FIRST = 20 -P_ARTIST_MORE = 40 -P_RELATED = 30 -P_SEARCH = 60 - -ACTIVE_WINDOW = 1800 # a user seen this recently counts as active -IDLE_DELAY = 20 # extra seconds between requests when nobody is active -MIN_POOL = 500 # crawl at full pace until the catalogue has this many tracks -REFRESH_AFTER = 86400 # re-check an artist's newest page for uploads this often - - -class Crawler: - def __init__(self, client, db): - self.client = client - self.db = db - self.queue = [] - self.queued = set() - self.seq = itertools.count() - self.cv = threading.Condition() - self.last_seed = 0.0 - # Logged-in pl8list sessions, kept in memory only while that user's feed is crawled. - self.feed_sessions = {} - - def push(self, priority, kind, arg): - key = (kind, arg) - with self.cv: - if key in self.queued: - return - self.queued.add(key) - heapq.heappush(self.queue, (priority, next(self.seq), kind, arg)) - self.cv.notify() - - def pop(self): - with self.cv: - while not self.queue: - self.cv.wait(timeout=30) - if not self.queue: - self._refill() - priority, _, kind, arg = heapq.heappop(self.queue) - self.queued.discard((kind, arg)) - return priority, kind, arg - - # Feedback hooks, called from the web server ---------------------------- - - def on_positive(self, track_id): - """Liked or listened through: crawl its related tracks and its artist's catalogue.""" - row = self.db.query("SELECT handle, slug FROM tracks WHERE id=?", (track_id,)) - if not row: - return - handle, slug = row[0]["handle"], row[0]["slug"] - self.push(P_FEEDBACK, "related", (track_id, handle, slug)) - self.push(P_FEEDBACK, "artist", handle) - - def on_login(self, user_id, session): - """Crawl the user's personal feed (artists in their pl8list library) with their session.""" - old = self.feed_sessions.get(user_id) - self.feed_sessions[user_id] = session - if old: - Client(old).logout() - self.push(P_FEEDBACK, "feed", (user_id, 1)) - - # Main loop ------------------------------------------------------------- - - def start(self): - threading.Thread(target=self.run, daemon=True, name="crawler").start() - - def run(self): - self._seed() - while True: - _, kind, arg = self.pop() - try: - getattr(self, "do_" + kind)(arg) - except Exception: - log.exception("crawl %s %r failed", kind, arg) - if kind != "feed" and not self._busy(): - time.sleep(IDLE_DELAY) - if time.time() - self.last_seed > 600: - self._seed() - - def _busy(self): - """Crawl at full pace while someone is listening or the catalogue is still small.""" - if self.db.active_since(time.time() - ACTIVE_WINDOW): - return True - return self.db.query("SELECT COUNT(*) AS n FROM tracks")[0]["n"] < MIN_POOL - - def _seed(self): - self.last_seed = time.time() - self.push(P_SEED, "home", None) - self.push(P_SEED, "ticker", None) - self.push(P_SEED, "best_artists", None) - for term in SEARCH_TERMS: - self.push(P_SEARCH, "search", term) - - def _refill(self): - """Queue ran dry: queue more artist pages, related lookups and stale artists.""" - for r in self.db.query(""" - SELECT handle FROM artists - WHERE total_pages IS NULL OR pages_fetched < total_pages - ORDER BY verified DESC, pages_fetched ASC LIMIT 20 - """): - self.push(P_ARTIST_MORE, "artist", r["handle"]) - for r in self.db.query(""" - SELECT id, handle, slug FROM tracks WHERE related_fetched=0 - ORDER BY saves + downloads DESC LIMIT 10 - """): - self.push(P_RELATED, "related", (r["id"], r["handle"], r["slug"])) - for r in self.db.query(""" - SELECT handle FROM artists - WHERE pages_fetched > 0 AND last_crawled < ? - ORDER BY last_crawled LIMIT 20 - """, (time.time() - REFRESH_AFTER,)): - self.push(P_ARTIST_MORE, "artist_refresh", r["handle"]) - - # Jobs ------------------------------------------------------------------ - - def _add_artist(self, handle, source, name=None, img=None, verified=0, priority=P_ARTIST_FIRST): - known = self.db.query("SELECT pages_fetched FROM artists WHERE handle=?", (handle,)) - self.db.upsert_artist(handle, name, img, verified, source) - if not known or known[0]["pages_fetched"] == 0: - self.push(priority, "artist", handle) - - def do_home(self, _): - d = self.client.home() or {} - for section in ("bestSellers", "recommended"): - self.db.upsert_tracks([normalize_item(i) for i in (d.get(section) or {}).get("items", [])]) - for e in d.get("tickerEvents") or []: - self._add_artist(e["artistHandle"], "ticker", e.get("artistName")) - - def do_ticker(self, _): - for e in (self.client.activity_ticker() or {}).get("events", []): - self._add_artist(e["artistHandle"], "ticker", e.get("artistName")) - - def do_best_artists(self, _): - d = self.client.best_selling_artists() or {} - for a in d.get("artists") or []: - self._add_artist(a["artist_handle"], "best_sellers", a.get("artist_name"), - a.get("artist_img"), a.get("artist_isVerified"), priority=P_SEED) - self.db.upsert_tracks([normalize_item(i) for i in (d.get("recommended") or {}).get("items", [])]) - - def do_feed(self, arg): - user_id, page = arg - session = self.feed_sessions.get(user_id) - if not session: - return - more = False - try: - d = Client(session).feed(page) or {} - self.db.upsert_tracks([normalize_item(i) for i in d.get("tracks") or []]) - for i in d.get("tracks") or []: - self.push(P_ARTIST_FIRST, "artist", i["track"]["user"]["handle"]) - more = page < (d.get("totalPages") or 0) - finally: - # Done or failed: forget the session here and end it at pl8list. - if more: - self.push(P_FEEDBACK, "feed", (user_id, page + 1)) - elif self.feed_sessions.pop(user_id, None): - Client(session).logout() - - def do_artist_refresh(self, handle): - """Newest uploads are on page 1, so re-reading it catches new tracks.""" - d = self.client.artist(handle, 1) - if d: - new = self.db.upsert_tracks([normalize_item(i) for i in d.get("tracks") or []]) - if new: - log.info("artist %s: %d new uploads", handle, new) - self.db.execute("UPDATE artists SET last_crawled=? WHERE handle=?", (time.time(), handle)) - - def do_artist(self, handle): - row = self.db.query("SELECT pages_fetched, total_pages FROM artists WHERE handle=?", (handle,)) - fetched = row[0]["pages_fetched"] if row else 0 - total = row[0]["total_pages"] if row else None - if total is not None and fetched >= total: - return - page = fetched + 1 - d = self.client.artist(handle, page) - if not d: - self.db.execute("UPDATE artists SET total_pages=0, last_crawled=? WHERE handle=?", - (time.time(), handle)) - return - new = self.db.upsert_tracks([normalize_item(i) for i in d.get("tracks") or []]) - total = d.get("totalPages") or 0 - self.db.execute("UPDATE artists SET pages_fetched=?, total_pages=?, last_crawled=? WHERE handle=?", - (page, total, time.time(), handle)) - log.info("artist %s page %d/%d: %d new tracks", handle, page, total, new) - if page < total: - self.push(P_ARTIST_MORE + page, "artist", handle) - - def do_related(self, arg): - track_id, handle, slug = arg - d = self.client.related(handle, slug) - self.db.execute("UPDATE tracks SET related_fetched=1 WHERE id=?", (track_id,)) - if not d: - return - rows = normalize_related(d) - self.db.upsert_tracks(rows, related_of=track_id) - for r in rows: - self._add_artist(r["handle"], "related", r["artist_name"], r["artist_img"], r["artist_verified"], - priority=P_RELATED) - - def do_search(self, term): - row = self.db.query("SELECT pages_fetched, total_pages FROM search_terms WHERE term=?", (term,)) - fetched = row[0]["pages_fetched"] if row else 0 - total = row[0]["total_pages"] if row else None - if total is not None and fetched >= total: - return - page = fetched + 1 - d = self.client.search_users(term, page) or {} - for item in d.get("items") or []: - a = item.get("artist") or {} - if a.get("handle"): - self._add_artist(a["handle"], "search", a.get("name"), a.get("img"), a.get("isVerified"), - priority=P_ARTIST_FIRST if a.get("isVerified") else P_SEARCH + 5) - total = d.get("totalPages") or 0 - self.db.execute( - "INSERT INTO search_terms (term, pages_fetched, total_pages) VALUES (?,?,?) " - "ON CONFLICT(term) DO UPDATE SET pages_fetched=excluded.pages_fetched, total_pages=excluded.total_pages", - (term, page, total)) - if page < total: - self.push(P_SEARCH + page, "search", term) diff --git a/db.py b/db.py deleted file mode 100644 index 4c2b1fb..0000000 --- a/db.py +++ /dev/null @@ -1,196 +0,0 @@ -"""SQLite store: the shared track catalogue, plus per-user accounts and listening events.""" - -import hashlib -import secrets -import sqlite3 -import threading -import time - -SESSION_DAYS = 90 - -SCHEMA = """ -CREATE TABLE IF NOT EXISTS artists ( - handle TEXT PRIMARY KEY, - name TEXT, - img TEXT, - verified INTEGER DEFAULT 0, - source TEXT, - pages_fetched INTEGER DEFAULT 0, - total_pages INTEGER, - last_crawled REAL -); -CREATE TABLE IF NOT EXISTS tracks ( - id INTEGER PRIMARY KEY, - handle TEXT NOT NULL, - artist_name TEXT, - artist_img TEXT, - artist_verified INTEGER DEFAULT 0, - slug TEXT NOT NULL, - title TEXT, - description TEXT, - bpm REAL, - key TEXT, - duration REAL, - preview_duration REAL, - price INTEGER DEFAULT 0, - created_at TEXT, - upload_id TEXT, - preview_upload_id TEXT, - waveform TEXT, - preview_waveform TEXT, - full_access INTEGER DEFAULT 0, - saves INTEGER DEFAULT 0, - downloads INTEGER DEFAULT 0, - purchases INTEGER DEFAULT 0, - comments INTEGER DEFAULT 0, - related_of INTEGER, - related_fetched INTEGER DEFAULT 0, - updated_at REAL -); -CREATE INDEX IF NOT EXISTS tracks_handle ON tracks(handle); -CREATE TABLE IF NOT EXISTS events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - track_id INTEGER NOT NULL, - kind TEXT NOT NULL, - seconds REAL, - ts REAL NOT NULL -); --- Users are pl8list accounts, keyed by their pl8list user id. No pl8list session is stored. -CREATE TABLE IF NOT EXISTS users ( - id INTEGER PRIMARY KEY, - handle TEXT NOT NULL, - name TEXT, - img TEXT, - created_at REAL NOT NULL, - last_seen REAL -); --- Our own login sessions. Only a hash of the cookie token is kept. -CREATE TABLE IF NOT EXISTS sessions ( - token_hash TEXT PRIMARY KEY, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - created_at REAL NOT NULL, - expires_at REAL NOT NULL -); -CREATE TABLE IF NOT EXISTS search_terms ( - term TEXT PRIMARY KEY, - pages_fetched INTEGER DEFAULT 0, - total_pages INTEGER -); -""" - -TRACK_COLS = [ - "id", "handle", "artist_name", "artist_img", "artist_verified", "slug", "title", "description", - "bpm", "key", "duration", "preview_duration", "price", "created_at", "upload_id", - "preview_upload_id", "waveform", "preview_waveform", "full_access", "saves", "downloads", - "purchases", "comments", -] - - -class DB: - def __init__(self, path): - self.conn = sqlite3.connect(path, check_same_thread=False, isolation_level=None) - self.conn.row_factory = sqlite3.Row - self.conn.execute("PRAGMA journal_mode=WAL") - self.conn.execute("PRAGMA foreign_keys=ON") - self.conn.executescript(SCHEMA) - # Databases from the single-user version have events without an owner. - if "user_id" not in [r["name"] for r in self.conn.execute("PRAGMA table_info(events)")]: - self.conn.execute("DELETE FROM events") - self.conn.execute("ALTER TABLE events ADD COLUMN user_id INTEGER") - self.conn.execute("CREATE INDEX IF NOT EXISTS events_user ON events(user_id, kind, track_id)") - self.lock = threading.Lock() - - def query(self, sql, args=()): - with self.lock: - return self.conn.execute(sql, args).fetchall() - - def execute(self, sql, args=()): - with self.lock: - return self.conn.execute(sql, args) - - def upsert_tracks(self, rows, related_of=None): - """Insert or refresh tracks. Returns how many were new.""" - new = 0 - now = time.time() - with self.lock: - for r in rows: - if not r: - continue - exists = self.conn.execute("SELECT 1 FROM tracks WHERE id=?", (r["id"],)).fetchone() - vals = [r[c] for c in TRACK_COLS] - if exists: - sets = ",".join("%s=?" % c for c in TRACK_COLS[1:]) - self.conn.execute("UPDATE tracks SET %s, updated_at=? WHERE id=?" % sets, - vals[1:] + [now, r["id"]]) - else: - new += 1 - self.conn.execute( - "INSERT INTO tracks (%s, related_of, updated_at) VALUES (%s)" - % (",".join(TRACK_COLS), ",".join("?" * (len(TRACK_COLS) + 2))), - vals + [related_of, now]) - self.conn.execute( - "INSERT INTO artists (handle, name, img, verified, source) VALUES (?,?,?,?,?) " - "ON CONFLICT(handle) DO UPDATE SET name=excluded.name, img=excluded.img, " - "verified=excluded.verified", - (r["handle"], r["artist_name"], r["artist_img"], r["artist_verified"], "track")) - return new - - def upsert_artist(self, handle, name=None, img=None, verified=0, source=None): - self.execute( - "INSERT INTO artists (handle, name, img, verified, source) VALUES (?,?,?,?,?) " - "ON CONFLICT(handle) DO UPDATE SET name=COALESCE(excluded.name, name), " - "img=COALESCE(excluded.img, img), verified=MAX(verified, excluded.verified)", - (handle, name, img, int(bool(verified)), source)) - - def add_event(self, user_id, track_id, kind, seconds=None): - self.execute("INSERT INTO events (user_id, track_id, kind, seconds, ts) VALUES (?,?,?,?,?)", - (user_id, track_id, kind, seconds, time.time())) - - def stats(self, user_id): - row = self.query(""" - SELECT (SELECT COUNT(*) FROM tracks) AS tracks, - (SELECT COUNT(*) FROM artists) AS artists, - (SELECT COUNT(*) FROM artists WHERE pages_fetched > 0) AS artists_crawled, - (SELECT COUNT(DISTINCT track_id) FROM events WHERE user_id=? AND kind='served') AS served - """, (user_id,))[0] - return dict(row) - - # Accounts -------------------------------------------------------------- - - def upsert_user(self, user): - now = time.time() - self.execute( - "INSERT INTO users (id, handle, name, img, created_at, last_seen) VALUES (?,?,?,?,?,?) " - "ON CONFLICT(id) DO UPDATE SET handle=excluded.handle, name=excluded.name, img=excluded.img, " - "last_seen=excluded.last_seen", - (user["id"], user["handle"], user.get("name"), user.get("img"), now, now)) - - def create_session(self, user_id): - """Returns the raw token for the cookie; only its hash is stored.""" - token = secrets.token_urlsafe(32) - now = time.time() - self.execute("INSERT INTO sessions (token_hash, user_id, created_at, expires_at) VALUES (?,?,?,?)", - (_hash(token), user_id, now, now + SESSION_DAYS * 86400)) - return token - - def session_user(self, token): - rows = self.query(""" - SELECT u.* FROM sessions s JOIN users u ON u.id = s.user_id - WHERE s.token_hash=? AND s.expires_at > ? - """, (_hash(token), time.time())) - if not rows: - return None - user = dict(rows[0]) - if (user["last_seen"] or 0) < time.time() - 60: - self.execute("UPDATE users SET last_seen=? WHERE id=?", (time.time(), user["id"])) - return user - - def delete_session(self, token): - self.execute("DELETE FROM sessions WHERE token_hash=?", (_hash(token),)) - - def active_since(self, ts): - return self.query("SELECT COUNT(*) AS n FROM users WHERE last_seen > ?", (ts,))[0]["n"] - - -def _hash(token): - return hashlib.sha256(token.encode()).hexdigest() diff --git a/pl8list.py b/pl8list.py deleted file mode 100644 index afbfd92..0000000 --- a/pl8list.py +++ /dev/null @@ -1,253 +0,0 @@ -"""Thin client for pl8list.com. - -pl8list is a Remix app, so every page loader can be read as JSON by adding -`?_data=` to the page URL. Resource routes under /api/ return JSON -directly. All clients share one throttle so we stay polite. - -Public data (artist pages, related tracks, previews, free tracks) works without -an account, so the crawler runs as a guest. A user's own session is only needed -for login and for reading their personal feed. -""" - -import base64 -import json -import re -import threading -import time -import urllib.error -import urllib.parse -import urllib.request - -BASE = "https://pl8list.com" -USER_AGENT = "pl8list-wrapper/0.2 (infinite-scroll client, https://pl8.narl.io)" - -_throttle_lock = threading.Lock() -_throttle_last = 0.0 - - -def _throttle(min_interval): - global _throttle_last - with _throttle_lock: - wait = _throttle_last + min_interval - time.monotonic() - if wait > 0: - time.sleep(wait) - _throttle_last = time.monotonic() - - -class Client: - def __init__(self, session_cookie=None, min_interval=0.7): - self.session_cookie = session_cookie - self.min_interval = min_interval - - def request(self, path, params=None, headers=None): - """Raw request. Returns the open response; caller must close it.""" - url = BASE + path - if params: - url += ("&" if "?" in url else "?") + urllib.parse.urlencode(params) - h = {"User-Agent": USER_AGENT, **(headers or {})} - if self.session_cookie: - h["Cookie"] = "_session=" + self.session_cookie - return urllib.request.urlopen(urllib.request.Request(url, headers=h), timeout=30) - - def get_json(self, path, params=None, route=None): - params = dict(params or {}) - if route: - params["_data"] = route - for attempt in range(4): - _throttle(self.min_interval) - try: - with self.request(path, params) as r: - body = r.read() - return json.loads(body) if body else None - except urllib.error.HTTPError as e: - if e.code in (429, 502, 503, 504) and attempt < 3: - time.sleep(2 ** attempt * 3) - continue - if e.code in (403, 404): - return None - raise - return None - - # Loaders ------------------------------------------------------------- - - def root(self): - return self.get_json("/", route="root") - - def home(self): - return self.get_json("/", route="routes/_index") - - # Paginated loaders ignore `page` unless `pagination=true` is also set, - # which is what the site's own useInfiniteLoading hook sends. - - def feed(self, page=1): - return self.get_json("/api/feed", {"page": page, "pagination": "true"}) - - def artist(self, handle, page=1): - return self.get_json("/" + urllib.parse.quote(handle), {"page": page, "pagination": "true"}, - route="routes/$handle") - - def related(self, handle, slug): - path = "/%s/%s/related" % (urllib.parse.quote(handle), urllib.parse.quote(slug)) - return self.get_json(path, route="routes/$handle_.$slug.related") - - def best_selling_artists(self): - return self.get_json("/best-selling-artists", route="routes/best-selling-artists") - - def activity_ticker(self): - return self.get_json("/api/activity-ticker") - - def search_users(self, q, page=1): - return self.get_json("/search/users", {"q": q, "page": page, "pagination": "true"}, - route="routes/search.users") - - def logout(self): - """End this session at pl8list. Errors are ignored: we drop the session either way.""" - try: - self.request("/logout", {"_data": "routes/logout"}).close() - except OSError: - pass - - -class LoginError(Exception): - pass - - -class LoginFlow: - """pl8list's passwordless login: POST email, pl8list mails a 6-digit code, POST code. - - The site keeps the pending login in its `_session` cookie, so one flow object - holds that cookie between the two steps. After `verify` succeeds, - `self.session` is the user's logged-in pl8list session. - """ - - def __init__(self, email): - self.email = email - self.session = None - - def _call(self, method, path, route, fields=None): - url = BASE + path + ("&" if "?" in path else "?") + urllib.parse.urlencode({"_data": route}) - headers = {"User-Agent": USER_AGENT} - if self.session: - headers["Cookie"] = "_session=" + self.session - data = None - if fields is not None: - data = urllib.parse.urlencode(fields).encode() - headers["Content-Type"] = "application/x-www-form-urlencoded" - req = urllib.request.Request(url, data=data, headers=headers, method=method) - try: - resp = urllib.request.urlopen(req, timeout=30) - except urllib.error.HTTPError as e: - resp = e - except OSError as e: - raise LoginError("pl8list unreachable: %s" % e) from e - with resp: - for c in resp.headers.get_all("Set-Cookie") or []: - m = re.match(r"_session=([^;]+)", c) - if m: - self.session = m.group(1) - body = resp.read() - status = resp.status if hasattr(resp, "status") else resp.code - try: - payload = json.loads(body) if body else None - except ValueError: - payload = None - return status, resp.headers, payload - - def _guest_id(self): - """The guest id pl8list stores in its cookie; the login forms echo it back.""" - try: - raw = urllib.parse.unquote(self.session).split(".")[0] - return json.loads(base64.b64decode(raw + "=" * (-len(raw) % 4))).get("pl8list_guest_id", "") - except (ValueError, AttributeError, TypeError): - return "" - - def _verify_path(self): - return "/verify?" + urllib.parse.urlencode({"email": self.email, "mode": "login"}) - - def start(self): - """Ask pl8list to mail a login code.""" - self._call("GET", "/login", "routes/login") - status, headers, payload = self._call("POST", "/login", "routes/login", - {"email": self.email, "pl8list_guest_id": self._guest_id()}) - if "/verify" not in (headers.get("X-Remix-Redirect") or ""): - raise LoginError((payload or {}).get("error") or "pl8list refused the login (HTTP %d)" % status) - - def resend(self): - _, _, payload = self._call("POST", self._verify_path(), "routes/verify", - {"action": "resend", "pl8list_guest_id": self._guest_id()}) - if payload and payload.get("resendError"): - raise LoginError(payload["resendError"]) - - def verify(self, code): - """Submit the code. Returns pl8list's currentUser for the new session.""" - _, _, payload = self._call("POST", self._verify_path(), "routes/verify", - {"action": "verify", "code": code, "pl8list_guest_id": self._guest_id()}) - if payload and payload.get("error"): - raise LoginError(payload["error"]) - user = (Client(self.session).root() or {}).get("currentUser") - if not user: - raise LoginError("login failed, try again") - if not user.get("handle"): - Client(self.session).logout() - self.session = None - raise LoginError("finish setting up your account on pl8list.com first (choose a handle)") - return user - - -def normalize_item(item): - """Flatten a loader track item ({track, hasFullAccess, saveCount, ...}) into a DB row.""" - return normalize_track(item["track"], { - "full_access": item.get("hasFullAccess"), - "saves": item.get("saveCount"), - "downloads": item.get("downloadCount"), - "purchases": item.get("purchaseCount"), - "comments": item.get("commentCount"), - }) - - -def normalize_related(data): - """The related loader returns bare tracks plus per-id maps for access and counts.""" - counts = data.get("counts") or {} - access = data.get("access") or {} - rows = [] - for t in data.get("tracks") or []: - k = str(t["id"]) - rows.append(normalize_track(t, { - "full_access": access.get(k), - "saves": (counts.get("saveCount") or {}).get(k), - "downloads": (counts.get("downloadCount") or {}).get(k), - "purchases": (counts.get("purchaseCount") or {}).get(k), - "comments": (counts.get("commentCount") or {}).get(k), - })) - return [r for r in rows if r] - - -def normalize_track(t, extra): - user = t.get("user") or {} - if not user.get("handle") or t.get("isDeleted") or not t.get("isPublic", True): - return None - return { - "id": t["id"], - "handle": user["handle"], - "artist_name": user.get("name") or user["handle"], - "artist_img": user.get("img"), - "artist_verified": int(bool(user.get("isVerified"))), - "slug": t["slug"], - "title": t.get("title") or t.get("filename") or t["slug"], - "description": t.get("description"), - "bpm": t.get("bpm"), - "key": t.get("key"), - "duration": t.get("duration"), - "preview_duration": t.get("previewDuration"), - "price": t.get("price") or 0, - "created_at": t.get("createdAt"), - "upload_id": t.get("uploadId"), - "preview_upload_id": t.get("previewUploadId"), - "waveform": json.dumps(t.get("waveform")) if t.get("waveform") else None, - "preview_waveform": json.dumps(t.get("previewWaveform")) if t.get("previewWaveform") else None, - "full_access": int(bool(extra.get("full_access"))), - "saves": extra.get("saves") or 0, - "downloads": extra.get("downloads") or 0, - "purchases": extra.get("purchases") or 0, - "comments": extra.get("comments") or 0, - } diff --git a/ranker.py b/ranker.py deleted file mode 100644 index f76e952..0000000 --- a/ranker.py +++ /dev/null @@ -1,166 +0,0 @@ -"""Pick the next batch of tracks for the scroll feed. - -Each unserved playable track gets a score: - - popularity log-scaled saves/downloads/purchases/comments, normalised to 0..1 - freshness exp(-age / 21 days) - artist learned from your events on that artist's tracks (likes, full listens, skips) - tempo / key closeness to the BPM and key of tracks you liked or finished - related track came from the "related" page of something you liked - exploration Gumbel noise, so the feed never freezes into one niche - -The batch is then picked greedily with a penalty for repeating an artist. -""" - -import datetime -import json -import math -import random -from collections import defaultdict - -WEIGHTS = { - "popularity": 1.2, - "freshness": 0.8, - "artist": 1.5, - "tempo": 1.0, - "key": 0.3, - "related": 1.0, - "unknown_artist": 0.2, - "exploration": 0.6, -} -# Per-event contributions to artist affinity. -EVENT_SIGNAL = {"like": 3.0, "unlike": -3.0, "complete": 1.5, "skip": -1.0, "dislike": -3.0} -REPEAT_ARTIST_PENALTY = 1.5 -RECENT_ARTIST_WINDOW = 6 - - -def _age_days(iso): - if not iso: - return 365.0 - try: - t = datetime.datetime.fromisoformat(iso.replace("Z", "+00:00")) - except ValueError: - return 365.0 - return max(0.0, (datetime.datetime.now(datetime.timezone.utc) - t).total_seconds() / 86400) - - -def _tempo_match(bpm, liked_bpms): - """Gaussian closeness to any liked tempo, also counting half/double time as a match.""" - if not bpm or not liked_bpms: - return 0.0 - best = 0.0 - for b in liked_bpms: - d = min(abs(bpm - b), abs(bpm * 2 - b), abs(bpm / 2 - b)) - best = max(best, math.exp(-((d / 6.0) ** 2))) - return best - - -def taste(db, user_id): - """Summarise a user's listening history into artist affinity, liked tempos/keys and liked track ids.""" - rows = db.query(""" - SELECT e.track_id, e.kind, e.seconds, t.handle, t.bpm, t.key - FROM events e JOIN tracks t ON t.id = e.track_id - WHERE e.user_id = ? AND e.kind IN ('like', 'unlike', 'complete', 'skip', 'dislike') - ORDER BY e.ts - """, (user_id,)) - artist = defaultdict(float) - liked = {} - completed = {} - for r in rows: - artist[r["handle"]] += EVENT_SIGNAL[r["kind"]] - if r["kind"] == "like": - liked[r["track_id"]] = r - elif r["kind"] == "unlike": - liked.pop(r["track_id"], None) - elif r["kind"] == "complete": - completed[r["track_id"]] = r - positive = {**completed, **liked} - return { - "artist": {h: math.tanh(v / 4.0) for h, v in artist.items()}, - "bpms": [r["bpm"] for r in positive.values() if r["bpm"]][-50:], - "keys": {r["key"] for r in positive.values() if r["key"]}, - "positive_ids": set(positive), - "liked_ids": set(liked), - } - - -def pick(db, user_id, n=10, free_only=False, bpm_min=None, bpm_max=None): - where = ["(t.full_access=1 OR t.preview_upload_id IS NOT NULL)", - "NOT EXISTS (SELECT 1 FROM events e WHERE e.user_id=? AND e.kind='served' AND e.track_id=t.id)"] - args = [user_id] - if free_only: - where.append("t.price = 0") - if bpm_min is not None: - where.append("t.bpm >= ?") - args.append(bpm_min) - if bpm_max is not None: - where.append("t.bpm <= ?") - args.append(bpm_max) - cands = db.query("SELECT * FROM tracks t WHERE " + " AND ".join(where), args) - if not cands: - return [] - - t = taste(db, user_id) - recent = [r["handle"] for r in db.query(""" - SELECT t.handle FROM events e JOIN tracks t ON t.id=e.track_id - WHERE e.user_id=? AND e.kind='served' ORDER BY e.ts DESC LIMIT ? - """, (user_id, RECENT_ARTIST_WINDOW))] - - def pop_raw(r): - return (math.log1p(r["saves"]) + 0.7 * math.log1p(r["downloads"]) - + 1.0 * math.log1p(r["purchases"]) + 0.5 * math.log1p(r["comments"])) - - max_pop = max(pop_raw(r) for r in cands) or 1.0 - w = WEIGHTS - scored = [] - for r in cands: - aff = t["artist"].get(r["handle"]) - s = (w["popularity"] * pop_raw(r) / max_pop - + w["freshness"] * math.exp(-_age_days(r["created_at"]) / 21.0) - + w["artist"] * (aff or 0.0) - + w["tempo"] * _tempo_match(r["bpm"], t["bpms"]) - + w["key"] * (1.0 if r["key"] in t["keys"] else 0.0) - + w["related"] * (1.0 if r["related_of"] in t["positive_ids"] else 0.0) - + w["unknown_artist"] * (1.0 if aff is None else 0.0) - + w["exploration"] * -math.log(-math.log(random.random() or 1e-12))) - scored.append([s, r]) - - out = [] - seen_artists = list(recent) - scored.sort(key=lambda x: -x[0]) - pool = scored[: max(n * 8, 60)] - while pool and len(out) < n: - best_i = max(range(len(pool)), - key=lambda i: pool[i][0] - REPEAT_ARTIST_PENALTY * seen_artists.count(pool[i][1]["handle"])) - s, r = pool.pop(best_i) - out.append(serialize(r, s)) - seen_artists.append(r["handle"]) - return out - - -def serialize(r, score=None): - preview = not r["full_access"] - wave = r["preview_waveform"] if preview and r["preview_waveform"] else r["waveform"] - return { - "id": r["id"], - "handle": r["handle"], - "slug": r["slug"], - "title": r["title"], - "description": r["description"], - "artist_name": r["artist_name"], - "artist_img": r["artist_img"], - "artist_verified": bool(r["artist_verified"]), - "bpm": r["bpm"], - "key": r["key"], - "duration": r["preview_duration"] if preview and r["preview_duration"] else r["duration"], - "price": r["price"], - "created_at": r["created_at"], - "saves": r["saves"], - "downloads": r["downloads"], - "purchases": r["purchases"], - "comments": r["comments"], - "preview": preview, - "upload_id": r["preview_upload_id"] if preview else r["upload_id"], - "waveform": json.loads(wave) if wave else None, - "score": round(score, 3) if score is not None else None, - } diff --git a/server.py b/server.py deleted file mode 100644 index a9a04c7..0000000 --- a/server.py +++ /dev/null @@ -1,442 +0,0 @@ -"""Infinite-scroll front end for pl8list, with sign-in through pl8list accounts. - - python3 server.py # then open http://127.0.0.1:8790 - -Users sign in with their pl8list email: pl8list mails them a code, they type it -here, and this server checks it with pl8list. Afterwards they get a session -cookie for this app only. Their pl8list session is used once to read their -personal feed and is never written to disk. - -Configuration comes from the environment (or .env): - - HOST bind address (default 127.0.0.1) - PORT port (default 8790) - PL8_DB SQLite path (default ./pl8list.db) - TRUST_PROXY 1 when behind a reverse proxy: trust X-Forwarded-For/-Proto -""" - -import argparse -import collections -import http.cookies -import json -import logging -import os -import re -import secrets -import threading -import time -import urllib.error -import urllib.parse -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from pathlib import Path - -import ranker -from crawler import Crawler -from db import DB, SESSION_DAYS -from pl8list import Client, LoginError, LoginFlow - -ROOT = Path(__file__).resolve().parent -STATIC = ROOT / "static" -EVENT_KINDS = {"play", "like", "unlike", "skip", "complete", "dislike"} -# Only HLS playlists and segments are proxied, nothing else on pl8list. -PROXY_RE = re.compile(r"^/api/[^/]+/[^/]+/hls/(playlist\.m3u8|segment/\d+)$") -EMAIL_RE = re.compile(r"^[^@\s]{1,64}@[^@\s]+\.[^@\s]+$") -CONTENT_TYPES = { - ".html": "text/html; charset=utf-8", ".js": "text/javascript", ".css": "text/css", - ".png": "image/png", ".svg": "image/svg+xml", ".webmanifest": "application/manifest+json", -} -SESSION_COOKIE = "pl8inf" -LOGIN_COOKIE = "pl8inf_login" -LOGIN_TTL = 900 -MAX_PENDING_LOGINS = 500 -MAX_RESENDS = 2 -MAX_CODE_ATTEMPTS = 5 -MAX_BODY = 4096 -CSP = "; ".join([ - "default-src 'self'", - "script-src 'self' https://cdn.jsdelivr.net", - "style-src 'self'", - "img-src 'self' data: https://pl8list.com https://files.pl8list.com", - "media-src 'self' blob:", - "worker-src 'self' blob:", - "connect-src 'self'", - "base-uri 'none'", - "form-action 'self'", - "frame-ancestors 'none'", -]) - -log = logging.getLogger("server") - - -def load_env(path): - if not path.exists(): - return - for line in path.read_text().splitlines(): - line = line.strip() - if line and not line.startswith("#") and "=" in line: - k, v = line.split("=", 1) - os.environ.setdefault(k.strip(), v.strip().strip("'\"")) - - -class RateLimiter: - """At most `limit` hits per key within `window` seconds.""" - - def __init__(self, limit, window): - self.limit, self.window = limit, window - self.hits = collections.defaultdict(collections.deque) - self.lock = threading.Lock() - - def allow(self, key): - now = time.time() - with self.lock: - if len(self.hits) > 10000: - for k in [k for k, q in self.hits.items() if not q or q[-1] < now - self.window]: - del self.hits[k] - q = self.hits[key] - while q and q[0] < now - self.window: - q.popleft() - if len(q) >= self.limit: - return False - q.append(now) - return True - - -class PendingLogins: - """Logins waiting for their emailed code, keyed by a random id in a short-lived cookie.""" - - def __init__(self): - self.flows = {} - self.lock = threading.Lock() - - def add(self, flow): - """Returns the new flow id, or None when too many logins are pending.""" - flow_id = secrets.token_urlsafe(24) - with self.lock: - now = time.time() - for k in [k for k, f in self.flows.items() if f.created < now - LOGIN_TTL]: - del self.flows[k] - if len(self.flows) >= MAX_PENDING_LOGINS: - return None - flow.created, flow.attempts, flow.resends = now, 0, 0 - self.flows[flow_id] = flow - return flow_id - - def count(self, flow, field, limit): - """Increment flow. under the lock. False once it goes over `limit`.""" - with self.lock: - setattr(flow, field, getattr(flow, field) + 1) - return getattr(flow, field) <= limit - - def get(self, flow_id): - with self.lock: - flow = self.flows.get(flow_id or "") - if flow and flow.created < time.time() - LOGIN_TTL: - del self.flows[flow_id] - return None - return flow - - def drop(self, flow_id): - with self.lock: - self.flows.pop(flow_id, None) - - -class Handler(BaseHTTPRequestHandler): - server_version = "pl8inf" - sys_version = "" - db: DB - client: Client - crawler: Crawler - trust_proxy = False - logins = PendingLogins() - login_ip_limit = RateLimiter(5, 900) - login_email_limit = RateLimiter(3, 900) - # Every start and resend makes pl8list send a mail; cap them for the whole server. - login_global_limit = RateLimiter(30, 600) - - def log_message(self, fmt, *args): - log.debug(fmt, *args) - - def end_headers(self): - self.send_header("Content-Security-Policy", CSP) - self.send_header("X-Content-Type-Options", "nosniff") - self.send_header("Referrer-Policy", "strict-origin-when-cross-origin") - self.send_header("X-Frame-Options", "DENY") - super().end_headers() - - # Request helpers ------------------------------------------------------ - - def client_ip(self): - # The proxy appends the address it saw, so only the last entry is trustworthy. - # Anything before it was sent by the client. - if self.trust_proxy and self.headers.get("X-Forwarded-For"): - return self.headers["X-Forwarded-For"].split(",")[-1].strip() - return self.client_address[0] - - def allow_login_mail(self, email): - return (self.login_ip_limit.allow(self.client_ip()) and self.login_email_limit.allow(email) - and self.login_global_limit.allow("")) - - def is_https(self): - return self.trust_proxy and self.headers.get("X-Forwarded-Proto") == "https" - - def cookie(self, name): - jar = http.cookies.SimpleCookie() - try: - jar.load(self.headers.get("Cookie") or "") - except http.cookies.CookieError: - return None - return jar[name].value if name in jar else None - - def set_cookie(self, name, value, max_age): - parts = ["%s=%s" % (name, value), "Path=/", "HttpOnly", "SameSite=Lax", "Max-Age=%d" % max_age] - if self.is_https(): - parts.append("Secure") - self.send_header("Set-Cookie", "; ".join(parts)) - - def current_user(self): - token = self.cookie(SESSION_COOKIE) - return self.db.session_user(token) if token else None - - def read_json(self): - length = int(self.headers.get("Content-Length") or 0) - if length > MAX_BODY: - raise ValueError("body too large") - data = json.loads(self.rfile.read(length) or b"{}") - if not isinstance(data, dict): - raise ValueError("expected an object") - return data - - def same_origin(self): - """Reject cross-site POSTs. Browsers always send Origin on POST.""" - origin = self.headers.get("Origin") - return not origin or urllib.parse.urlsplit(origin).netloc == self.headers.get("Host") - - def send_json(self, obj, status=200, cookies=()): - body = json.dumps(obj).encode() - self.send_response(status) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.send_header("Cache-Control", "no-store") - for c in cookies: - self.set_cookie(*c) - self.end_headers() - self.wfile.write(body) - - def send_static(self, name): - path = (STATIC / name).resolve() - if STATIC not in path.parents or not path.is_file(): - return self.send_error(404) - body = path.read_bytes() - self.send_response(200) - self.send_header("Content-Type", CONTENT_TYPES.get(path.suffix, "application/octet-stream")) - self.send_header("Content-Length", str(len(body))) - self.send_header("Cache-Control", "no-cache") - self.end_headers() - self.wfile.write(body) - - # Routing -------------------------------------------------------------- - - def do_GET(self): - url = urllib.parse.urlsplit(self.path) - if url.path == "/": - return self.send_static("index.html") - if url.path.startswith("/static/"): - return self.send_static(url.path[len("/static/"):]) - if url.path == "/healthz": - return self.send_json({"ok": True}) - user = self.current_user() - if url.path == "/local/me": - if not user: - return self.send_json({"error": "not signed in"}, 401) - return self.send_json({"user": {k: user[k] for k in ("handle", "name", "img")}}) - if not user: - return self.send_error(401) - if url.path == "/local/feed": - return self.feed(user, urllib.parse.parse_qs(url.query)) - if url.path == "/local/likes": - return self.likes(user) - if PROXY_RE.match(url.path): - return self.proxy(url) - self.send_error(404) - - def do_POST(self): - if not self.same_origin(): - return self.send_error(403) - try: - data = self.read_json() - except ValueError: - return self.send_error(400) - routes = { - "/auth/start": self.auth_start, - "/auth/resend": self.auth_resend, - "/auth/verify": self.auth_verify, - "/auth/logout": self.auth_logout, - "/local/event": self.event, - } - handler = routes.get(self.path) - if not handler: - return self.send_error(404) - handler(data) - - # Auth ----------------------------------------------------------------- - - def auth_start(self, data): - email = str(data.get("email") or "").strip().lower() - if len(email) > 254 or not EMAIL_RE.match(email): - return self.send_json({"error": "enter a valid email"}, 400) - if not self.allow_login_mail(email): - return self.send_json({"error": "too many attempts, wait a few minutes"}, 429) - flow = LoginFlow(email) - flow_id = self.logins.add(flow) - if not flow_id: - return self.send_json({"error": "too many sign-ins right now, try again in a few minutes"}, 429) - try: - flow.start() - except LoginError as e: - self.logins.drop(flow_id) - return self.send_json({"error": str(e)}, 400) - self.send_json({"ok": True, "email": email}, cookies=[(LOGIN_COOKIE, flow_id, LOGIN_TTL)]) - - def auth_resend(self, data): - flow = self.logins.get(self.cookie(LOGIN_COOKIE)) - if not flow: - return self.send_json({"error": "login expired, start again"}, 400) - if not self.logins.count(flow, "resends", MAX_RESENDS): - return self.send_json({"error": "no more resends, start again"}, 429) - if not self.allow_login_mail(flow.email): - return self.send_json({"error": "too many attempts, wait a few minutes"}, 429) - try: - flow.resend() - except LoginError as e: - return self.send_json({"error": str(e)}, 400) - self.send_json({"ok": True}) - - def auth_verify(self, data): - flow_id = self.cookie(LOGIN_COOKIE) - flow = self.logins.get(flow_id) - if not flow: - return self.send_json({"error": "login expired, start again"}, 400) - code = str(data.get("code") or "").strip() - if not re.fullmatch(r"\d{6}", code): - return self.send_json({"error": "the code has 6 digits"}, 400) - if not self.logins.count(flow, "attempts", MAX_CODE_ATTEMPTS): - self.logins.drop(flow_id) - return self.send_json({"error": "too many wrong codes, start again"}, 429) - try: - pl8_user = flow.verify(code) - except LoginError as e: - return self.send_json({"error": str(e)}, 400) - self.logins.drop(flow_id) - self.db.upsert_user(pl8_user) - self.crawler.on_login(pl8_user["id"], flow.session) - token = self.db.create_session(pl8_user["id"]) - log.info("signed in: %s", pl8_user["handle"]) - self.send_json({"user": {"handle": pl8_user["handle"], "name": pl8_user.get("name")}}, - cookies=[(SESSION_COOKIE, token, SESSION_DAYS * 86400), (LOGIN_COOKIE, "", 0)]) - - def auth_logout(self, data): - token = self.cookie(SESSION_COOKIE) - if token: - self.db.delete_session(token) - self.send_json({"ok": True}, cookies=[(SESSION_COOKIE, "", 0)]) - - # App ------------------------------------------------------------------ - - def event(self, data): - user = self.current_user() - if not user: - return self.send_error(401) - try: - track_id, kind = int(data["track_id"]), data["kind"] - seconds = float(data["seconds"]) if data.get("seconds") is not None else None - except (ValueError, KeyError, TypeError): - return self.send_error(400) - if kind not in EVENT_KINDS: - return self.send_error(400) - self.db.add_event(user["id"], track_id, kind, seconds) - if kind in ("like", "complete"): - self.crawler.on_positive(track_id) - self.send_json({"ok": True}) - - def feed(self, user, q): - def num(name): - try: - return float(q[name][0]) if q.get(name, [""])[0] else None - except ValueError: - return None - - n = max(1, min(int(num("n") or 10), 30)) - kwargs = dict(free_only=q.get("free", [""])[0] == "1", bpm_min=num("bpm_min"), bpm_max=num("bpm_max")) - # On a cold start the crawler needs a few seconds before there is anything to serve. - deadline = time.time() + 20 - tracks = ranker.pick(self.db, user["id"], n, **kwargs) - while not tracks and time.time() < deadline: - time.sleep(1.5) - tracks = ranker.pick(self.db, user["id"], n, **kwargs) - for t in tracks: - self.db.add_event(user["id"], t["id"], "served") - stats = self.db.stats(user["id"]) - stats["liked"] = len(ranker.taste(self.db, user["id"])["liked_ids"]) - self.send_json({"tracks": tracks, "stats": stats}) - - def likes(self, user): - liked = list(ranker.taste(self.db, user["id"])["liked_ids"]) - rows = self.db.query(""" - SELECT t.*, MAX(e.ts) AS liked_at FROM events e JOIN tracks t ON t.id = e.track_id - WHERE e.user_id = ? AND e.kind = 'like' AND t.id IN (%s) - GROUP BY t.id ORDER BY liked_at DESC - """ % ",".join("?" * len(liked)), [user["id"]] + liked) if liked else [] - self.send_json({"tracks": [ranker.serialize(r) for r in rows]}) - - def proxy(self, url): - # Previews and free tracks stream fine as a guest. Segments redirect to - # signed storage URLs; urllib follows that server-side. - try: - upstream = self.client.request(url.path, urllib.parse.parse_qsl(url.query)) - except urllib.error.HTTPError as e: - return self.send_error(e.code) - except OSError: - return self.send_error(502) - with upstream: - self.send_response(200) - self.send_header("Content-Type", upstream.headers.get("Content-Type", "application/octet-stream")) - if upstream.headers.get("Content-Length"): - self.send_header("Content-Length", upstream.headers["Content-Length"]) - self.send_header("Cache-Control", "private, max-age=600") - self.end_headers() - try: - while chunk := upstream.read(64 * 1024): - self.wfile.write(chunk) - except (BrokenPipeError, ConnectionResetError): - pass - - -def main(): - load_env(ROOT / ".env") - ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--host", default=os.environ.get("HOST", "127.0.0.1")) - ap.add_argument("--port", type=int, default=int(os.environ.get("PORT", "8790"))) - ap.add_argument("--db", default=os.environ.get("PL8_DB", str(ROOT / "pl8list.db"))) - ap.add_argument("-v", "--verbose", action="store_true") - args = ap.parse_args() - logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, - format="%(asctime)s %(name)s %(message)s", datefmt="%H:%M:%S") - - db = DB(args.db) - client = Client() - crawler = Crawler(client, db) - crawler.start() - - Handler.db, Handler.client, Handler.crawler = db, client, crawler - Handler.trust_proxy = os.environ.get("TRUST_PROXY") == "1" - server = ThreadingHTTPServer((args.host, args.port), Handler) - server.daemon_threads = True - log.info("listening on http://%s:%d", args.host, args.port) - try: - server.serve_forever() - except KeyboardInterrupt: - pass - - -if __name__ == "__main__": - main()