Add opt-in pl8list connection with library sync, follows and resync

At sign-in users can choose to stay connected to pl8list on that device.
The pl8list session is then stored in the device's session row, encrypted
with XChaCha20-Poly1305 under a key derived from the device's local token
(only the browser has it) and a server key kept outside the database. It
is only decrypted during that device's own requests, re-encrypted when the
token rotates or pl8list sends a new cookie, and deleted when pl8list
rejects it.

Every sign-in now syncs the personal feed, saved tracks (imported as likes)
and saved artists (imported as follows). Connected devices resync on use
every six hours and push new likes and follows to the pl8list library with
addOnly, so nothing is ever removed there. Without the connection, the
resync button runs the email-code sign-in again.

Artists can be followed from each card (f); follows feed the ranker. The
account menu shows the sync state, resync and disconnect. Privacy policy,
export and account deletion cover the new data.

backend/src/config.rs

@@ -16,6 +16,10 @@ pub struct Config {
     pub secure: bool,
     pub cache_dir: PathBuf,
     pub cache_max_bytes: u64,
+    /// Server half of the key for stored pl8list sessions: `LINK_KEY` (base64, 32
+    /// bytes), else the file `LINK_KEY_FILE` (created on first start).
+    pub link_key: Option<String>,
+    pub link_key_file: PathBuf,
     /// pl8list itself; only changed by tests.
     pub pl8_base: String,
     pub legal: Legal,
@@ -65,6 +69,9 @@ pub fn load_dotenv(path: &str) {
 impl Config {
     pub fn from_env() -> Self {
         let db_path = PathBuf::from(var("PL8_DB").unwrap_or_else(|| "pl8list.db".into()));
+        let link_key_file = var("LINK_KEY_FILE").map(PathBuf::from).unwrap_or_else(|| {
+            db_path.parent().map(|p| p.join("link.key")).unwrap_or_else(|| PathBuf::from("link.key"))
+        });
         let cache_dir = var("CACHE_DIR").map(PathBuf::from).unwrap_or_else(|| {
             db_path.parent().map(|p| p.join("cache")).unwrap_or_else(|| PathBuf::from("cache"))
         });
@@ -81,6 +88,8 @@ impl Config {
             secure: flag("INSECURE_COOKIES") != Some(true),
             cache_dir,
             cache_max_bytes: var("CACHE_MAX_MB").and_then(|v| v.parse().ok()).unwrap_or(2048u64) << 20,
+            link_key: var("LINK_KEY"),
+            link_key_file,
             pl8_base: var("PL8_BASE").unwrap_or_else(|| "https://pl8list.com".into()),
             legal: Legal {
                 name: var("IMPRESSUM_NAME").unwrap_or_default(),

backend/src/crawler.rs

@@ -9,8 +9,8 @@
 //! 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.
 //!
-//! A user's pl8list session never enters the queue: `crawl_feed` owns it in its own
-//! task, reads the personal feed, logs the session out at pl8list and drops it.
+//! A user's pl8list session never enters the queue; account sync (`sync.rs`) reads
+//! with it and only hands the artists it found to `discovered_artists`.
 
 use std::{
     cmp::Reverse,
@@ -19,7 +19,6 @@ use std::{
     time::Duration,
 };
 
-use secrecy::SecretString;
 use serde_json::Value;
 use tokio::{sync::Notify, time::Instant};
 
@@ -48,8 +47,6 @@ const IDLE_DELAY: Duration = Duration::from_secs(20); // extra wait between requ
 const MIN_POOL: i64 = 500; // crawl at full pace until the catalogue has this many tracks
 const REFRESH_AFTER: f64 = 86400.0; // re-check an artist's newest page for uploads this often
 const RESEED_EVERY: Duration = Duration::from_secs(600);
-const FEED_MAX_PAGES: i64 = 20;
-const FEED_DEADLINE: Duration = Duration::from_secs(120);
 
 #[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug)]
 enum Job {
@@ -140,37 +137,10 @@ impl Crawler {
         }
     }
 
-    /// Read the user's personal feed (artists in their pl8list library) with their
-    /// freshly logged-in session, then end that session at pl8list. Takes ownership
-    /// of the session; it is dropped (and zeroed) when this task ends.
-    pub fn crawl_feed(self: &Arc<Self>, session: SecretString) {
-        let me = self.clone();
-        tokio::spawn(async move {
-            let res = tokio::time::timeout(FEED_DEADLINE, me.read_feed(&session)).await;
-            match res {
-                Ok(Ok(pages)) => tracing::info!("personal feed: {pages} pages"),
-                Ok(Err(e)) => tracing::warn!("personal feed failed: {e:#}"),
-                Err(_) => tracing::warn!("personal feed timed out"),
-            }
-            me.pl8.logout(&session).await;
-        });
-    }
-
-    async fn read_feed(&self, session: &SecretString) -> anyhow::Result<i64> {
-        let mut page = 1;
-        loop {
-            let d = self.pl8.feed(session, page).await?.unwrap_or(Value::Null);
-            let list = d.get("tracks");
-            let rows = normalize::items(list);
-            for r in &rows {
-                self.push(P_ARTIST_FIRST, Job::Artist(r.handle.clone()));
-            }
-            self.db.call(move |c| db::upsert_tracks(c, &rows, None)).await?;
-            let total = d.get("totalPages").and_then(Value::as_i64).unwrap_or(0);
-            if page >= total.min(FEED_MAX_PAGES) {
-                return Ok(page);
-            }
-            page += 1;
+    /// Artists found in a user's pl8list feed or library: crawl them soon.
+    pub fn discovered_artists(&self, handles: impl IntoIterator<Item = String>) {
+        for h in handles {
+            self.push(P_ARTIST_FIRST, Job::Artist(h));
         }
     }
 

backend/src/db.rs

@@ -93,6 +93,20 @@ CREATE INDEX IF NOT EXISTS sessions_user ON sessions(user_id);
 CREATE INDEX IF NOT EXISTS events_kind_ts ON events(kind, ts);
 "#;
 
+// pl8_link: the device's pl8list session, encrypted (see link.rs). NULL unless the
+// user chose to stay connected on that device.
+const SCHEMA_V3: &str = r#"
+ALTER TABLE sessions ADD COLUMN pl8_link BLOB;
+ALTER TABLE sessions ADD COLUMN pl8_synced_at REAL;
+CREATE TABLE IF NOT EXISTS follows (
+    user_id INTEGER NOT NULL,
+    handle TEXT NOT NULL,
+    source TEXT NOT NULL,
+    ts REAL NOT NULL,
+    PRIMARY KEY (user_id, handle)
+);
+"#;
+
 pub fn now() -> f64 {
     SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64()
 }
@@ -145,6 +159,9 @@ fn migrate(conn: &Connection) -> rusqlite::Result<()> {
         conn.execute_batch(SCHEMA_V2)?;
         conn.pragma_update(None, "user_version", 2)?;
     }
+    if version < 3 {
+        conn.execute_batch(&format!("BEGIN; {SCHEMA_V3} PRAGMA user_version = 3; COMMIT;"))?;
+    }
     Ok(())
 }
 
@@ -314,6 +331,10 @@ pub fn add_event(conn: &Connection, user_id: i64, track_id: i64, kind: &str, sec
     Ok(())
 }
 
+pub fn track_path(conn: &Connection, track_id: i64) -> rusqlite::Result<Option<(String, String)>> {
+    conn.query_row("SELECT handle, slug FROM tracks WHERE id=?", [track_id], |r| Ok((r.get(0)?, r.get(1)?))).optional()
+}
+
 pub fn track_count(conn: &Connection) -> rusqlite::Result<i64> {
     conn.query_row("SELECT COUNT(*) FROM tracks", [], |r| r.get(0))
 }
@@ -399,16 +420,26 @@ pub struct SessionHit {
     pub user: User,
     pub created_at: f64,
     pub expires_at: f64,
+    pub link: Option<Vec<u8>>,
+    pub synced_at: Option<f64>,
 }
 
 pub fn session_user(conn: &Connection, token: &str) -> rusqlite::Result<Option<SessionHit>> {
     let now = now();
     let hit = conn
         .query_row(
-            "SELECT u.*, s.created_at AS s_created, s.expires_at AS s_expires \
+            "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 > ?",
             params![hash_token(token), now],
-            |r| Ok(SessionHit { user: User::from_row(r)?, created_at: r.get("s_created")?, expires_at: r.get("s_expires")? }),
+            |r| {
+                Ok(SessionHit {
+                    user: User::from_row(r)?,
+                    created_at: r.get("s_created")?,
+                    expires_at: r.get("s_expires")?,
+                    link: r.get("pl8_link")?,
+                    synced_at: r.get("pl8_synced_at")?,
+                })
+            },
         )
         .optional()?;
     if let Some(h) = &hit {
@@ -437,6 +468,25 @@ pub fn rotate_session(conn: &Connection, old_token: &str, user_id: i64) -> rusql
     Ok(token)
 }
 
+/// Store (or replace) the encrypted pl8list session of one device.
+pub fn set_link(conn: &Connection, token_hash: &str, blob: &[u8], synced_at: Option<f64>) -> rusqlite::Result<()> {
+    conn.execute(
+        "UPDATE sessions SET pl8_link=?, pl8_synced_at=COALESCE(?, pl8_synced_at) WHERE token_hash=?",
+        params![blob, synced_at, token_hash],
+    )?;
+    Ok(())
+}
+
+pub fn clear_link(conn: &Connection, token_hash: &str) -> rusqlite::Result<()> {
+    conn.execute("UPDATE sessions SET pl8_link=NULL, pl8_synced_at=NULL WHERE token_hash=?", [token_hash])?;
+    Ok(())
+}
+
+pub fn mark_synced(conn: &Connection, token_hash: &str) -> rusqlite::Result<()> {
+    conn.execute("UPDATE sessions SET pl8_synced_at=? WHERE token_hash=? AND pl8_link IS NOT NULL", params![now(), token_hash])?;
+    Ok(())
+}
+
 pub fn delete_session(conn: &Connection, token: &str) -> rusqlite::Result<()> {
     conn.execute("DELETE FROM sessions WHERE token_hash=?", [hash_token(token)])?;
     Ok(())
@@ -447,6 +497,50 @@ pub fn delete_user_sessions(conn: &Connection, user_id: i64) -> rusqlite::Result
     Ok(())
 }
 
+// Library sync ---------------------------------------------------------------
+
+/// Tracks saved on pl8list count as likes, unless the user already liked or unliked
+/// them here. Returns how many were added.
+pub fn import_likes(conn: &mut Connection, user_id: i64, track_ids: &[i64]) -> rusqlite::Result<usize> {
+    let tx = conn.transaction()?;
+    let mut added = 0;
+    {
+        let mut insert = tx.prepare_cached(
+            "INSERT INTO events (user_id, track_id, kind, seconds, ts) SELECT ?1, ?2, 'like', NULL, ?3 \
+             WHERE NOT EXISTS (SELECT 1 FROM events WHERE user_id=?1 AND track_id=?2 AND kind IN ('like', 'unlike'))",
+        )?;
+        let now = now();
+        for id in track_ids {
+            added += insert.execute(params![user_id, id, now])?;
+        }
+    }
+    tx.commit()?;
+    Ok(added)
+}
+
+pub fn add_follows(conn: &mut Connection, user_id: i64, handles: &[String], source: &str) -> rusqlite::Result<usize> {
+    let tx = conn.transaction()?;
+    let mut added = 0;
+    {
+        let mut insert = tx.prepare_cached("INSERT OR IGNORE INTO follows (user_id, handle, source, ts) VALUES (?,?,?,?)")?;
+        let now = now();
+        for h in handles {
+            added += insert.execute(params![user_id, h, source, now])?;
+        }
+    }
+    tx.commit()?;
+    Ok(added)
+}
+
+pub fn unfollow(conn: &Connection, user_id: i64, handle: &str) -> rusqlite::Result<()> {
+    conn.execute("DELETE FROM follows WHERE user_id=? AND handle=?", params![user_id, handle])?;
+    Ok(())
+}
+
+pub fn follows(conn: &Connection, user_id: i64) -> rusqlite::Result<std::collections::HashSet<String>> {
+    conn.prepare_cached("SELECT handle FROM follows WHERE user_id=?")?.query_map([user_id], |r| r.get(0))?.collect()
+}
+
 pub fn active_since(conn: &Connection, ts: f64) -> rusqlite::Result<i64> {
     conn.query_row("SELECT COUNT(*) FROM users WHERE last_seen > ?", [ts], |r| r.get(0))
 }
@@ -455,9 +549,17 @@ pub fn active_since(conn: &Connection, ts: f64) -> rusqlite::Result<i64> {
 pub fn export_user(conn: &Connection, user_id: i64) -> rusqlite::Result<serde_json::Value> {
     let user = conn.query_row("SELECT * FROM users WHERE id=?", [user_id], User::from_row)?;
     let sessions: Vec<serde_json::Value> = conn
-        .prepare("SELECT created_at, expires_at FROM sessions WHERE user_id=? ORDER BY created_at")?
+        .prepare(
+            "SELECT created_at, expires_at, pl8_link IS NOT NULL, pl8_synced_at FROM sessions \
+             WHERE user_id=? ORDER BY created_at",
+        )?
         .query_map([user_id], |r| {
-            Ok(serde_json::json!({ "created_at": r.get::<_, f64>(0)?, "expires_at": r.get::<_, f64>(1)? }))
+            Ok(serde_json::json!({
+                "created_at": r.get::<_, f64>(0)?,
+                "expires_at": r.get::<_, f64>(1)?,
+                "pl8list_connected": r.get::<_, bool>(2)?,
+                "pl8list_synced_at": r.get::<_, Option<f64>>(3)?,
+            }))
         })?
         .collect::<Result<_, _>>()?;
     let events: Vec<serde_json::Value> = conn
@@ -477,12 +579,21 @@ pub fn export_user(conn: &Connection, user_id: i64) -> rusqlite::Result<serde_js
             }))
         })?
         .collect::<Result<_, _>>()?;
+    let follows: Vec<serde_json::Value> = conn
+        .prepare("SELECT handle, source, ts FROM follows WHERE user_id=? ORDER BY ts")?
+        .query_map([user_id], |r| {
+            Ok(serde_json::json!({ "artist": r.get::<_, String>(0)?, "source": r.get::<_, String>(1)?, "ts": r.get::<_, f64>(2)? }))
+        })?
+        .collect::<Result<_, _>>()?;
     Ok(serde_json::json!({
         "exported_at": now(),
         "user": user,
         "sessions": sessions,
+        "follows": follows,
         "events": events,
-        "note": "Times are Unix timestamps in seconds. Session tokens are stored only as hashes and are not included.",
+        "note": "Times are Unix timestamps in seconds. Session tokens are stored only as hashes and are not included. \
+                 A connected session's pl8list login is stored encrypted with a key only your browser holds, so it \
+                 cannot be included either.",
     }))
 }
 
@@ -490,6 +601,7 @@ pub fn export_user(conn: &Connection, user_id: i64) -> rusqlite::Result<serde_js
 pub fn delete_user(conn: &mut Connection, user_id: i64) -> rusqlite::Result<()> {
     let tx = conn.transaction()?;
     tx.execute("DELETE FROM events WHERE user_id=?", [user_id])?;
+    tx.execute("DELETE FROM follows WHERE user_id=?", [user_id])?;
     tx.execute("DELETE FROM sessions WHERE user_id=?", [user_id])?;
     tx.execute("DELETE FROM users WHERE id=?", [user_id])?;
     tx.commit()
@@ -551,6 +663,19 @@ mod tests {
         assert_eq!(count("SELECT COUNT(*) FROM events WHERE user_id=2"), 1);
     }
 
+    #[test]
+    fn imported_likes_do_not_override_choices_made_here() {
+        let mut c = mem();
+        upsert_user(&c, 1, "a", None, None).unwrap();
+        add_event(&c, 1, 10, "unlike", None).unwrap();
+        assert_eq!(import_likes(&mut c, 1, &[10, 11]).unwrap(), 1);
+        assert_eq!(import_likes(&mut c, 1, &[10, 11]).unwrap(), 0);
+        assert_eq!(add_follows(&mut c, 1, &["x".into(), "y".into()], "pl8list").unwrap(), 2);
+        assert_eq!(add_follows(&mut c, 1, &["x".into()], "local").unwrap(), 0);
+        unfollow(&c, 1, "x").unwrap();
+        assert_eq!(follows(&c, 1).unwrap().into_iter().collect::<Vec<_>>(), vec!["y".to_string()]);
+    }
+
     #[test]
     fn upsert_counts_new_tracks() {
         let mut c = mem();

backend/src/http/account.rs

@@ -1,4 +1,5 @@
-//! Data export, account deletion and signing out everywhere (Art. 15, 17, 20 GDPR).
+//! Data export, account deletion and signing out everywhere (Art. 15, 17, 20 GDPR),
+//! plus the pl8list connection of this device.
 
 use axum::{
     Extension, Json,
@@ -8,7 +9,7 @@ use axum::{
 };
 use serde_json::{Value, json};
 
-use super::{CurrentUser, Res, St, auth::signed_out, json_error};
+use super::{CurrentUser, Res, St, auth::signed_out, json_error, no_store};
 use crate::db;
 
 pub async fn export(State(app): St, Extension(current): Extension<CurrentUser>) -> Res {
@@ -41,6 +42,11 @@ pub async fn delete(State(app): St, Extension(current): Extension<CurrentUser>,
         return Ok(json_error(StatusCode::BAD_REQUEST, "type your handle to confirm"));
     }
     let uid = user.id;
+    // Other devices' stored sessions can't be opened without their cookies; deleting
+    // the rows makes them unusable. This device's one is logged out at pl8list too.
+    if let Some((_, session)) = app.connected(&current).await {
+        app.pl8.logout(&session).await;
+    }
     app.db.call(move |c| db::delete_user(c, uid)).await?;
     tracing::info!("account deleted");
     Ok(signed_out(&app, json!({ "ok": true })))
@@ -51,6 +57,40 @@ pub async fn logout_all(State(app): St, Extension(current): Extension<CurrentUse
         Ok(u) => u.id,
         Err(r) => return Ok(r),
     };
+    if let Some((_, session)) = app.connected(&current).await {
+        app.pl8.logout(&session).await;
+    }
     app.db.call(move |c| db::delete_user_sessions(c, uid)).await?;
     Ok(signed_out(&app, json!({ "ok": true })))
 }
+
+/// The resync button. A connected device syncs with its stored session; any other
+/// device has to sign in with pl8list again (409 tells the page to show the sign-in).
+pub async fn sync(State(app): St, Extension(current): Extension<CurrentUser>) -> Res {
+    let uid = match current.require() {
+        Ok(u) => u.id,
+        Err(r) => return Ok(r),
+    };
+    let Some((sealer, session)) = app.connected(&current).await else {
+        return Ok(json_error(StatusCode::CONFLICT, "sign in with pl8list again to sync"));
+    };
+    if !app.limits.resync.allow(&uid.to_string()) {
+        return Ok(json_error(StatusCode::TOO_MANY_REQUESTS, "synced a moment ago, try again in two minutes"));
+    }
+    let started = app.sync.linked(sealer, session);
+    Ok((no_store(), Json(json!({ "ok": true, "started": started }))).into_response())
+}
+
+/// Stop staying connected on this device: log the session out at pl8list and delete it.
+pub async fn disconnect(State(app): St, Extension(current): Extension<CurrentUser>) -> Res {
+    if let Err(r) = current.require() {
+        return Ok(r);
+    }
+    if let Some((_, session)) = app.connected(&current).await {
+        app.pl8.logout(&session).await;
+    }
+    if let Some(token) = current.token.clone() {
+        app.db.call(move |c| db::clear_link(c, &db::hash_token(&token))).await?;
+    }
+    Ok((no_store(), Json(json!({ "ok": true }))).into_response())
+}

backend/src/http/app.rs

@@ -1,4 +1,4 @@
-//! The feed, likes and listening events.
+//! The feed, likes, follows and listening events.
 
 use std::{collections::HashMap, time::Duration};
 
@@ -11,8 +11,8 @@ use axum::{
 use serde_json::{Value, json};
 use tokio::time::Instant;
 
-use super::{CurrentUser, Res, St, json_error, no_store};
-use crate::{db, ranker};
+use super::{App, CurrentUser, Res, St, json_error, no_store};
+use crate::{db, ranker, sync};
 
 const EVENT_KINDS: &[&str] = &["play", "like", "unlike", "skip", "complete", "dislike"];
 /// On a cold start the crawler needs a few seconds before there is anything to serve.
@@ -20,11 +20,38 @@ const FEED_WAIT: Duration = Duration::from_secs(10);
 
 pub async fn me(Extension(current): Extension<CurrentUser>) -> Response {
     match current.user {
-        Some(u) => (no_store(), Json(json!({ "user": { "handle": u.handle, "name": u.name, "img": u.img } }))).into_response(),
+        Some(u) => (
+            no_store(),
+            Json(json!({
+                "user": { "handle": u.handle, "name": u.name, "img": u.img },
+                "pl8list": { "connected": current.link.is_some(), "synced_at": current.synced_at },
+            })),
+        )
+            .into_response(),
         None => json_error(StatusCode::UNAUTHORIZED, "not signed in"),
     }
 }
 
+/// Mark tracks by artists the user follows.
+async fn mark_followed(app: &App, uid: i64, tracks: &mut [Value]) -> anyhow::Result<()> {
+    let follows = app.db.call(move |c| db::follows(c, uid)).await?;
+    for t in tracks {
+        let followed = t["handle"].as_str().is_some_and(|h| follows.contains(h));
+        t["followed"] = json!(followed);
+    }
+    Ok(())
+}
+
+/// A connected device resyncs in the background once its last sync is old enough.
+async fn resync_if_due(app: &App, current: &CurrentUser) {
+    if current.link.is_none() || current.synced_at.is_some_and(|t| t > db::now() - sync::RESYNC_AFTER) {
+        return;
+    }
+    if let Some((sealer, session)) = app.connected(current).await {
+        app.sync.linked(sealer, session);
+    }
+}
+
 fn num(q: &HashMap<String, String>, name: &str) -> Option<f64> {
     q.get(name).and_then(|v| v.parse::<f64>().ok()).filter(|v| v.is_finite())
 }
@@ -43,15 +70,17 @@ pub async fn feed(State(app): St, Extension(current): Extension<CurrentUser>, Qu
         bpm_min: num(&q, "bpm_min"),
         bpm_max: num(&q, "bpm_max"),
     };
+    resync_if_due(&app, &current).await;
     let deadline = Instant::now() + FEED_WAIT;
     let uid = user.id;
-    let tracks = loop {
+    let mut tracks = loop {
         let tracks = app.db.call(move |c| ranker::pick(c, uid, n, filters, ranker)).await?;
         if !tracks.is_empty() || Instant::now() >= deadline {
             break tracks;
         }
         tokio::time::sleep(Duration::from_millis(1500)).await;
     };
+    mark_followed(&app, uid, &mut tracks).await?;
     let ids: Vec<i64> = tracks.iter().filter_map(|t| t["id"].as_i64()).collect();
     let stats = app
         .db
@@ -72,7 +101,8 @@ pub async fn likes(State(app): St, Extension(current): Extension<CurrentUser>) -
         Ok(u) => u.id,
         Err(r) => return Ok(r),
     };
-    let tracks = app.db.call(move |c| ranker::likes(c, uid)).await?;
+    let mut tracks = app.db.call(move |c| ranker::likes(c, uid)).await?;
+    mark_followed(&app, uid, &mut tracks).await?;
     Ok((no_store(), Json(json!({ "tracks": tracks }))).into_response())
 }
 
@@ -99,5 +129,45 @@ pub async fn event(State(app): St, Extension(current): Extension<CurrentUser>, J
     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(&current).await) {
+            app.sync.push_like(sealer, session, handle, slug);
+        }
+    }
     Ok((no_store(), Json(json!({ "ok": true }))).into_response())
 }
+
+/// Valid pl8list handles, as seen in every handle crawled so far.
+fn valid_handle(h: &str) -> bool {
+    (1..=64).contains(&h.len()) && h.bytes().all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b)) && h != "." && h != ".."
+}
+
+/// Follow or unfollow an artist. Unfollowing only affects this app; on a connected
+/// device a follow is also saved to the user's pl8list library.
+pub async fn follow(State(app): St, Extension(current): Extension<CurrentUser>, Json(body): Json<Value>) -> Res {
+    let uid = match current.require() {
+        Ok(u) => u.id,
+        Err(r) => return Ok(r),
+    };
+    let handle = body.get("handle").and_then(Value::as_str).filter(|h| valid_handle(h)).map(String::from);
+    let on = body.get("follow").and_then(Value::as_bool);
+    let (Some(handle), Some(on)) = (handle, on) else {
+        return Ok(StatusCode::BAD_REQUEST.into_response());
+    };
+    if !app.limits.events.allow(&uid.to_string()) {
+        return Ok(StatusCode::TOO_MANY_REQUESTS.into_response());
+    }
+    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) })
+        .await?;
+    if added > 0 {
+        app.crawler.discovered_artists([handle.clone()]);
+        if app.limits.pl8_writes.allow(&uid.to_string()) && let Some((sealer, session)) = app.connected(&current).await {
+            app.sync.push_follow(sealer, session, handle);
+        }
+    }
+    Ok((no_store(), Json(json!({ "ok": true, "followed": on }))).into_response())
+}

backend/src/http/auth.rs

@@ -135,7 +135,7 @@ pub async fn resend(State(app): St, peer: Peer, headers: HeaderMap) -> Res {
     Ok((no_store(), Json(json!({ "ok": true }))).into_response())
 }
 
-pub async fn verify(State(app): St, headers: HeaderMap, Json(body): Json<Value>) -> Res {
+pub async fn verify(State(app): St, Extension(current): Extension<CurrentUser>, headers: HeaderMap, Json(body): Json<Value>) -> Res {
     let id = login_cookie(&app, &headers).map(String::from);
     let Some(pending) = app.logins.get(id.as_deref()) else { return Ok(expired()) };
     let id = id.unwrap_or_default();
@@ -153,18 +153,36 @@ pub async fn verify(State(app): St, headers: HeaderMap, Json(body): Json<Value>)
         Err(e) => return Ok(json_error(StatusCode::BAD_REQUEST, &e.0)),
     };
     app.logins.drop(&id);
-    // The pl8list session goes straight to the feed crawl, which logs it out afterwards.
-    app.crawler.crawl_feed(pl8_session);
+    // "Stay connected" is opt-in per device; without it the session is used for one
+    // sync and then logged out at pl8list.
+    let stay = body.get("connect").and_then(Value::as_bool) == Some(true);
     let u = user.clone();
+    // Signing in again on a device (the resync button) replaces its old session.
+    let previous = current.token.clone();
+    if let Some(old) = app.connected(&current).await {
+        app.pl8.logout(&old.1).await;
+    }
     let token = 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())?;
             db::create_session(c, u.id)
         })
         .await?;
-    tracing::info!("signed in: {}", user.handle);
-    let mut res = (no_store(), Json(json!({ "user": { "handle": user.handle, "name": user.name } }))).into_response();
+    if stay {
+        let sealer = app.sync.key.sealer(&token, user.id);
+        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);
+    } else {
+        app.sync.once(user.id, pl8_session);
+    }
+    tracing::info!("signed in: {}{}", user.handle, if stay { " (stays connected)" } else { "" });
+    let mut res = (no_store(), Json(json!({ "user": { "handle": user.handle, "name": user.name }, "connected": stay }))).into_response();
     let h = res.headers_mut();
     h.append(header::SET_COOKIE, app.cookies.set(SESSION_COOKIE, &token, (db::SESSION_IDLE_DAYS * 86400.0) as i64));
     h.append(header::SET_COOKIE, app.cookies.clear(LOGIN_COOKIE));
@@ -172,6 +190,9 @@ pub async fn verify(State(app): St, headers: HeaderMap, Json(body): Json<Value>)
 }
 
 pub async fn logout(State(app): St, Extension(current): Extension<CurrentUser>) -> Res {
+    if let Some((_, session)) = app.connected(&current).await {
+        app.pl8.logout(&session).await;
+    }
     if let Some(token) = current.token {
         app.db.call(move |c| db::delete_session(c, &token)).await?;
     }

backend/src/http/mod.rs

@@ -24,13 +24,17 @@ use serde_json::json;
 use tower::limit::GlobalConcurrencyLimitLayer;
 use tower_http::timeout::TimeoutLayer;
 
+use secrecy::SecretString;
+
 use crate::{
     cache::Cache,
     config::Config,
     crawler::Crawler,
     db::{self, Db, User},
+    link::{LinkKey, Sealer},
     pl8::Pl8,
     ratelimit::RateLimiter,
+    sync::Syncer,
 };
 use security::Cookies;
 
@@ -46,6 +50,10 @@ pub struct Limits {
     pub login_global: RateLimiter,
     pub events: RateLimiter,
     pub proxy: RateLimiter,
+    /// Saves to a user's pl8list library (likes and follows pushed from here).
+    pub pl8_writes: RateLimiter,
+    /// The resync button.
+    pub resync: RateLimiter,
 }
 
 pub struct App {
@@ -53,6 +61,7 @@ pub struct App {
     pub db: Db,
     pub pl8: Arc<Pl8>,
     pub crawler: Arc<Crawler>,
+    pub sync: Arc<Syncer>,
     pub cache: Cache,
     pub logins: auth::PendingLogins,
     pub limits: Limits,
@@ -64,8 +73,9 @@ pub struct App {
 pub type St = State<Arc<App>>;
 
 impl App {
-    pub fn new(cfg: Config, db: Db, pl8: Arc<Pl8>, crawler: Arc<Crawler>) -> Arc<Self> {
+    pub fn new(cfg: Config, db: Db, pl8: Arc<Pl8>, crawler: Arc<Crawler>, link: LinkKey) -> Arc<Self> {
         let cache = Cache::open(cfg.cache_dir.clone(), cfg.cache_max_bytes);
+        let sync = Syncer::new(pl8.clone(), db.clone(), crawler.clone(), link);
         let legal = legal::Pages::render(&cfg.legal);
         Arc::new(App {
             cookies: Cookies { secure: cfg.secure },
@@ -75,9 +85,12 @@ impl App {
                 login_global: RateLimiter::new(30, 600),
                 events: RateLimiter::new(600, 600),
                 proxy: RateLimiter::new(1200, 60),
+                pl8_writes: RateLimiter::new(60, 600),
+                resync: RateLimiter::new(1, 120),
             },
             logins: auth::PendingLogins::default(),
             segments: proxy::SegmentKeys::default(),
+            sync,
             cache,
             legal,
             cfg,
@@ -97,6 +110,21 @@ impl App {
         self.logins.prune();
         self.segments.prune();
     }
+
+    /// This device's stored pl8list session, if it stays connected. A blob that no
+    /// longer opens (for example after the server key changed) is dropped.
+    pub async fn connected(&self, current: &CurrentUser) -> Option<(Sealer, SecretString)> {
+        let (Some(user), Some(token), Some(blob)) = (&current.user, &current.token, &current.link) else { return None };
+        let sealer = self.sync.key.sealer(token, user.id);
+        match sealer.open(blob) {
+            Some(session) => Some((sealer, session)),
+            None => {
+                let hash = sealer.token_hash.clone();
+                let _ = self.db.call(move |c| db::clear_link(c, &hash)).await;
+                None
+            }
+        }
+    }
 }
 
 // Errors ------------------------------------------------------------------
@@ -133,6 +161,9 @@ pub fn no_store() -> [(header::HeaderName, &'static str); 1] {
 pub struct CurrentUser {
     pub user: Option<User>,
     pub token: Option<String>,
+    /// Encrypted pl8list session of this device (see `link.rs`).
+    pub link: Option<Vec<u8>>,
+    pub synced_at: Option<f64>,
 }
 
 impl CurrentUser {
@@ -152,14 +183,41 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response {
         None => (None, false),
     };
     let mut set_cookie: Vec<HeaderValue> = vec![];
-    let mut current = CurrentUser { user: None, token: token.clone() };
+    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 {
             let mut new_token = None;
+            current.link = hit.link.clone();
+            current.synced_at = hit.synced_at;
             if hit.created_at < db::now() - ROTATE_AFTER {
                 let (old, uid) = (t.clone(), hit.user.id);
-                new_token = app.db.call(move |c| db::rotate_session(c, &old, uid)).await.ok();
+                // The stored pl8list session is encrypted for the old token; move it over.
+                let reopened = hit.link.as_deref().and_then(|b| app.sync.key.sealer(&t, uid).open(b));
+                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) => {
+                        if let Some(session) = reopened {
+                            let new = key.sealer(&nt, uid);
+                            let blob = new.seal(&session);
+                            let old_hash = db::hash_token(&t);
+                            let stored = blob.clone();
+                            let _ = app
+                                .db
+                                .call(move |c| {
+                                    db::set_link(c, &new.token_hash, &stored, synced)?;
+                                    db::clear_link(c, &old_hash)
+                                })
+                                .await;
+                            current.link = Some(blob);
+                        } else {
+                            current.link = None;
+                        }
+                        Some(nt)
+                    }
+                    Err(_) => None,
+                };
             } else if legacy {
                 new_token = Some(t.clone());
             }
@@ -256,9 +314,12 @@ pub fn router(app: Arc<App>) -> Router {
         .route("/local/feed", get(app::feed))
         .route("/local/likes", get(app::likes))
         .route("/local/event", post(app::event))
+        .route("/local/follow", post(app::follow))
         .route("/account/export", get(account::export))
         .route("/account/delete", post(account::delete))
         .route("/account/logout-all", post(account::logout_all))
+        .route("/account/pl8list/sync", post(account::sync))
+        .route("/account/pl8list/disconnect", post(account::disconnect))
         .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))

backend/src/http/tests.rs

@@ -16,7 +16,7 @@ use wiremock::{
 };
 
 use super::{App, router};
-use crate::{config::{Config, Legal}, crawler::Crawler, db::Db, pl8::Pl8};
+use crate::{config::{Config, Legal}, crawler::Crawler, db::Db, link::LinkKey, pl8::Pl8};
 
 const LOGGED_IN: &str = "LOGGED-IN-PL8-SESSION-SECRET";
 
@@ -38,13 +38,15 @@ async fn setup(name: &str, pl8_base: &str) -> (Router, Arc<App>, PathBuf) {
         secure: true,
         cache_dir: db_path.parent().unwrap().join("cache"),
         cache_max_bytes: 1 << 20,
+        link_key: None,
+        link_key_file: db_path.parent().unwrap().join("link.key"),
         pl8_base: pl8_base.into(),
         legal: Legal::default(),
     };
     let db = Db::open(&db_path).unwrap();
     let pl8 = Arc::new(Pl8::new(pl8_base, Duration::ZERO).unwrap());
     let crawler = Crawler::new(pl8.clone(), db.clone());
-    let app = App::new(cfg, db, pl8, crawler);
+    let app = App::new(cfg, db, pl8, crawler, LinkKey::fixed(9));
     (router(app.clone()), app, db_path)
 }
 
@@ -222,3 +224,159 @@ async fn legacy_cookie_moves_to_host_prefix() {
     assert!(cookies.iter().any(|c| c.starts_with(&format!("__Host-pl8inf={token};"))));
     assert!(cookies.iter().any(|c| c.starts_with("pl8inf=;") && c.contains("Max-Age=0")));
 }
+
+async fn wait_for(pl8: &MockServer, what: impl Fn(&wiremock::Request) -> bool) -> bool {
+    for _ in 0..50 {
+        if pl8.received_requests().await.unwrap().iter().any(&what) {
+            return true;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    false
+}
+
+async fn sign_in(r: &Router, body: &str) -> (String, serde_json::Value) {
+    let res = r.clone().oneshot(req("POST", "/auth/start", Some(r#"{"email":"a@b.de"}"#), None, "203.0.113.9")).await.unwrap();
+    let login = cookie_pair(&set_cookies(&res), "__Host-pl8inf_login");
+    let res = r.clone().oneshot(req("POST", "/auth/verify", Some(body), Some(&login), "203.0.113.9")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::OK);
+    let session = cookie_pair(&set_cookies(&res), "__Host-pl8inf");
+    let bytes = axum::body::to_bytes(res.into_body(), 1 << 16).await.unwrap();
+    (session, serde_json::from_slice(&bytes).unwrap())
+}
+
+async fn json_body(res: axum::response::Response) -> serde_json::Value {
+    serde_json::from_slice(&axum::body::to_bytes(res.into_body(), 1 << 20).await.unwrap()).unwrap()
+}
+
+const REFRESHED: &str = "REFRESHED-PL8-SESSION-SECRET";
+
+#[tokio::test]
+async fn connected_device_syncs_library_and_pushes_likes_and_follows() {
+    use wiremock::matchers::path_regex;
+    let pl8 = fake_pl8list(0).await;
+    // pl8list hands out a new cookie with the feed; later calls must use it.
+    Mock::given(method("GET")).and(path("/api/feed"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(r#"{"tracks":[],"totalPages":1}"#)
+            .insert_header("set-cookie", format!("_session={REFRESHED}; Max-Age=7776000; Path=/").as_str()))
+        .with_priority(1)
+        .mount(&pl8).await;
+    Mock::given(method("GET")).and(path("/library/tracks")).and(query_param("_data", "routes/library.tracks"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(
+            r#"{"items":[{"track":{"id":77,"slug":"song","title":"Song","user":{"handle":"artist1"}}}],"totalPages":1}"#))
+        .mount(&pl8).await;
+    Mock::given(method("GET")).and(path("/library/artists")).and(query_param("_data", "routes/library.artists"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(r#"{"items":[{"handle":"artist2"}],"totalPages":1}"#))
+        .mount(&pl8).await;
+    Mock::given(method("POST")).and(path_regex(r"^/api/.+/add-to-library$")).and(query_param("addOnly", "true"))
+        .and(has_header("cookie", format!("_session={REFRESHED}").as_str()))
+        .respond_with(ResponseTemplate::new(200).set_body_string(r#"{"success":true,"wasInLibrary":false}"#))
+        .expect(2)
+        .mount(&pl8).await;
+    Mock::given(method("GET")).and(path("/logout"))
+        .and(has_header("cookie", format!("_session={REFRESHED}").as_str()))
+        .respond_with(ResponseTemplate::new(204))
+        .expect(1)
+        .mount(&pl8).await;
+    let (r, app, db_path) = setup("connected", &pl8.uri()).await;
+
+    let (session, body) = sign_in(&r, r#"{"code":"123456","connect":true}"#).await;
+    assert_eq!(body["connected"], true);
+    assert!(wait_for(&pl8, |q| q.url.path() == "/library/artists").await, "library not synced");
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    let me = json_body(r.clone().oneshot(req("GET", "/local/me", None, Some(&session), "203.0.113.9")).await.unwrap()).await;
+    assert_eq!(me["pl8list"]["connected"], true);
+    // Saved tracks became likes, saved artists follows.
+    let likes = json_body(r.clone().oneshot(req("GET", "/local/likes", None, Some(&session), "203.0.113.9")).await.unwrap()).await;
+    assert_eq!(likes["tracks"][0]["id"], 77);
+    let follows = app.db.call(|c| crate::db::follows(c, 42)).await.unwrap();
+    assert!(follows.contains("artist2"));
+
+    // The stored session is encrypted: neither cookie appears in the files.
+    let raw = [std::fs::read(&db_path).unwrap(), std::fs::read(db_path.with_extension("db-wal")).unwrap_or_default()].concat();
+    for secret in [LOGGED_IN, REFRESHED] {
+        assert!(!raw.windows(secret.len()).any(|w| w == secret.as_bytes()), "{secret} stored in plain text");
+    }
+
+    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);
+    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);
+    assert!(wait_for(&pl8, |q| q.url.path() == "/api/artist3/add-to-library").await);
+
+    // Resync works once, then is rate limited.
+    let res = r.clone().oneshot(req("POST", "/account/pl8list/sync", Some("{}"), Some(&session), "203.0.113.9")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::OK);
+    let res = r.clone().oneshot(req("POST", "/account/pl8list/sync", Some("{}"), Some(&session), "203.0.113.9")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS);
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    let res = r.clone().oneshot(req("POST", "/account/pl8list/disconnect", Some("{}"), Some(&session), "203.0.113.9")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::OK);
+    let me = json_body(r.clone().oneshot(req("GET", "/local/me", None, Some(&session), "203.0.113.9")).await.unwrap()).await;
+    assert_eq!(me["pl8list"]["connected"], false);
+    let res = r.clone().oneshot(req("POST", "/account/pl8list/sync", Some("{}"), Some(&session), "203.0.113.9")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::CONFLICT);
+    pl8.verify().await;
+}
+
+#[tokio::test]
+async fn expired_pl8list_session_disconnects_the_device() {
+    let pl8 = fake_pl8list(0).await;
+    Mock::given(method("GET")).and(path("/api/feed"))
+        .respond_with(ResponseTemplate::new(302).insert_header("location", "/login?continue=%2Fapi%2Ffeed"))
+        .with_priority(1)
+        .mount(&pl8).await;
+    let (r, _app, _) = setup("expired", &pl8.uri()).await;
+    let (session, _) = sign_in(&r, r#"{"code":"123456","connect":true}"#).await;
+    let mut connected = true;
+    for _ in 0..50 {
+        let me = json_body(r.clone().oneshot(req("GET", "/local/me", None, Some(&session), "203.0.113.9")).await.unwrap()).await;
+        connected = me["pl8list"]["connected"].as_bool().unwrap();
+        if !connected {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    assert!(!connected, "expired session still stored");
+}
+
+#[tokio::test]
+async fn token_rotation_keeps_the_pl8list_connection() {
+    let (r, app, _) = setup("rotate", "http://127.0.0.1:9").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 blob = app.sync.key.sealer(&token, 5).seal(&secrecy::SecretString::from("PL8".to_string()));
+    let hash = crate::db::hash_token(&token);
+    app.db
+        .call(move |c| {
+            crate::db::set_link(c, &hash, &blob, Some(crate::db::now()))?;
+            c.execute("UPDATE sessions SET created_at = created_at - 8 * 86400", [])
+        })
+        .await
+        .unwrap();
+    let res = r.oneshot(req("GET", "/local/me", None, Some(&format!("__Host-pl8inf={token}")), "203.0.113.1")).await.unwrap();
+    let new = cookie_pair(&set_cookies(&res), "__Host-pl8inf").trim_start_matches("__Host-pl8inf=").to_string();
+    assert_ne!(new, token);
+    assert_eq!(json_body(res).await["pl8list"]["connected"], true);
+    let (new_hash, old_hash) = (crate::db::hash_token(&new), crate::db::hash_token(&token));
+    let (new_blob, old_blob) = app
+        .db
+        .call(move |c| {
+            let get = |h: &str| c.query_row("SELECT pl8_link FROM sessions WHERE token_hash=?", [h], |r| r.get::<_, Option<Vec<u8>>>(0));
+            Ok((get(&new_hash)?, get(&old_hash)?))
+        })
+        .await
+        .unwrap();
+    use secrecy::ExposeSecret;
+    assert_eq!(app.sync.key.sealer(&new, 5).open(&new_blob.unwrap()).unwrap().expose_secret(), "PL8");
+    assert!(old_blob.is_none());
+}

backend/src/link.rs

@@ -0,0 +1,153 @@
+//! Opt-in "stay connected to pl8list": a device keeps its pl8list session, encrypted.
+//!
+//! Each stored session is encrypted with a key derived from two secrets:
+//!
+//! - the device's local session token, which only the browser holds (the database
+//!   has just its SHA-256 hash), and
+//! - a server key kept outside the database (`LINK_KEY`, or a key file).
+//!
+//! A copy of the database, even together with the server key, opens nothing, and a
+//! stolen browser cookie only reaches our API. The server can decrypt a session only
+//! while that device is making a request; background jobs never can.
+
+use std::{fs, io::Write, path::Path};
+
+use base64::Engine;
+use chacha20poly1305::{
+    XChaCha20Poly1305, XNonce,
+    aead::{Aead, KeyInit, Payload},
+};
+use hkdf::Hkdf;
+use rand::Rng;
+use secrecy::{ExposeSecret, SecretString};
+use sha2::Sha256;
+use zeroize::Zeroizing;
+
+use crate::db;
+
+const VERSION: u8 = 1;
+const INFO: &[u8] = b"pl8inf pl8list link v1";
+const NONCE_LEN: usize = 24;
+
+pub struct LinkKey(Zeroizing<[u8; 32]>);
+
+impl LinkKey {
+    /// `LINK_KEY` (base64, 32 bytes) if set, else the key file, created on first start.
+    pub fn load(env: Option<&str>, file: &Path) -> anyhow::Result<Self> {
+        let b64 = base64::engine::general_purpose::STANDARD;
+        let decode = |s: &str| -> anyhow::Result<Self> {
+            let raw = Zeroizing::new(b64.decode(s.trim())?);
+            let key: [u8; 32] = raw.as_slice().try_into().map_err(|_| anyhow::anyhow!("LINK_KEY must be 32 bytes"))?;
+            Ok(LinkKey(Zeroizing::new(key)))
+        };
+        if let Some(s) = env.filter(|s| !s.trim().is_empty()) {
+            return decode(s);
+        }
+        if let Ok(s) = fs::read_to_string(file) {
+            return decode(&s);
+        }
+        let mut key = Zeroizing::new([0u8; 32]);
+        rand::rng().fill_bytes(key.as_mut());
+        let mut opts = fs::OpenOptions::new();
+        opts.write(true).create_new(true);
+        #[cfg(unix)]
+        std::os::unix::fs::OpenOptionsExt::mode(&mut opts, 0o600);
+        opts.open(file)?.write_all(b64.encode(key.as_ref()).as_bytes())?;
+        tracing::info!("created link key {}", file.display());
+        Ok(LinkKey(key))
+    }
+
+    #[cfg(test)]
+    pub fn fixed(byte: u8) -> Self {
+        LinkKey(Zeroizing::new([byte; 32]))
+    }
+
+    /// Everything needed to seal and open one device's pl8list session. Holds the
+    /// derived key, not the token.
+    pub fn sealer(&self, token: &str, user_id: i64) -> Sealer {
+        let mut key = Zeroizing::new([0u8; 32]);
+        Hkdf::<Sha256>::new(Some(self.0.as_ref()), token.as_bytes())
+            .expand(INFO, key.as_mut())
+            .expect("32 bytes is a valid HKDF-SHA256 output length");
+        Sealer { key, user_id, token_hash: db::hash_token(token) }
+    }
+}
+
+pub struct Sealer {
+    key: Zeroizing<[u8; 32]>,
+    pub user_id: i64,
+    pub token_hash: String,
+}
+
+impl Sealer {
+    /// Binds the ciphertext to this user and this session row.
+    fn aad(&self) -> Vec<u8> {
+        let mut aad = self.user_id.to_le_bytes().to_vec();
+        aad.extend_from_slice(self.token_hash.as_bytes());
+        aad
+    }
+
+    fn cipher(&self) -> XChaCha20Poly1305 {
+        XChaCha20Poly1305::new_from_slice(self.key.as_ref()).expect("key is 32 bytes")
+    }
+
+    pub fn seal(&self, session: &SecretString) -> Vec<u8> {
+        let mut nonce = [0u8; NONCE_LEN];
+        rand::rng().fill_bytes(&mut nonce);
+        let aad = self.aad();
+        let sealed = self
+            .cipher()
+            .encrypt(&XNonce::from(nonce), Payload { msg: session.expose_secret().as_bytes(), aad: &aad })
+            .expect("encryption with a valid key does not fail");
+        let mut out = Vec::with_capacity(1 + NONCE_LEN + sealed.len());
+        out.push(VERSION);
+        out.extend_from_slice(&nonce);
+        out.extend_from_slice(&sealed);
+        out
+    }
+
+    pub fn open(&self, blob: &[u8]) -> Option<SecretString> {
+        let (&version, rest) = blob.split_first()?;
+        if version != VERSION || rest.len() < NONCE_LEN {
+            return None;
+        }
+        let (nonce, sealed) = rest.split_at(NONCE_LEN);
+        let nonce: [u8; NONCE_LEN] = nonce.try_into().ok()?;
+        let aad = self.aad();
+        let plain = Zeroizing::new(self.cipher().decrypt(&XNonce::from(nonce), Payload { msg: sealed, aad: &aad }).ok()?);
+        let text = std::str::from_utf8(&plain).ok()?;
+        Some(SecretString::from(text.to_string()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn opens_only_with_the_same_token_key_and_user() {
+        let key = LinkKey::fixed(1);
+        let s = key.sealer("token-a", 7);
+        let blob = s.seal(&SecretString::from("cookie-value".to_string()));
+        assert!(!blob.windows(12).any(|w| w == b"cookie-value"));
+        assert_eq!(s.open(&blob).unwrap().expose_secret(), "cookie-value");
+        assert!(key.sealer("token-b", 7).open(&blob).is_none());
+        assert!(key.sealer("token-a", 8).open(&blob).is_none());
+        assert!(LinkKey::fixed(2).sealer("token-a", 7).open(&blob).is_none());
+        let mut tampered = blob.clone();
+        *tampered.last_mut().unwrap() ^= 1;
+        assert!(s.open(&tampered).is_none());
+    }
+
+    #[test]
+    fn key_file_is_created_once() {
+        let dir = std::env::temp_dir().join(format!("pl8inf-link-{}", std::process::id()));
+        fs::create_dir_all(&dir).unwrap();
+        let file = dir.join("link.key");
+        let _ = fs::remove_file(&file);
+        let a = LinkKey::load(None, &file).unwrap();
+        let b = LinkKey::load(None, &file).unwrap();
+        assert_eq!(a.0.as_ref(), b.0.as_ref());
+        fs::remove_dir_all(&dir).unwrap();
+    }
+}

backend/src/main.rs

@@ -5,8 +5,10 @@
 //!
 //! 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, then logged out at pl8list; it is never written to disk.
+//! cookie for this app only. Their pl8list session is used once to sync their
+//! library, then logged out at pl8list, unless they opt in to stay connected on
+//! that device; then it is stored encrypted with a key only their browser can
+//! unlock (see `link.rs`).
 //! See `config.rs` for the environment variables.
 
 #![allow(clippy::result_large_err)] // handlers return axum `Response` as their error
@@ -16,9 +18,11 @@ mod config;
 mod crawler;
 mod db;
 mod http;
+mod link;
 mod pl8;
 mod ranker;
 mod ratelimit;
+mod sync;
 
 use std::{
     io::{Read, Write},
@@ -71,12 +75,13 @@ async fn main() -> anyhow::Result<std::process::ExitCode> {
         tracing::warn!("INSECURE_COOKIES=1: cookies without Secure, no HSTS. Use only for local development.");
     }
     let db = db::Db::open(&cfg.db_path)?;
+    let link = link::LinkKey::load(cfg.link_key.as_deref(), &cfg.link_key_file)?;
     let pl8 = Arc::new(pl8::Pl8::new(&cfg.pl8_base, Duration::from_millis(700))?);
     let crawler = crawler::Crawler::new(pl8.clone(), db.clone());
     crawler.start();
 
     let addr: SocketAddr = format!("{}:{}", cfg.host, cfg.port).parse()?;
-    let app = http::App::new(cfg, db, pl8, crawler);
+    let app = http::App::new(cfg, db, pl8, crawler, link);
     {
         let app = app.clone();
         tokio::spawn(async move {

backend/src/pl8/login.rs

@@ -3,7 +3,8 @@
 //! The site keeps the pending login in its `_session` cookie, so one flow holds that
 //! cookie between the two steps. After `verify` succeeds the cookie is a fully
 //! logged-in pl8list session. It is handed to the caller exactly once and is never
-//! logged, stored or kept in the flow.
+//! logged or kept in the flow; only an opted-in device stores it, encrypted
+//! (see `link.rs`).
 
 use base64::{
     Engine,
@@ -14,7 +15,7 @@ use percent_encoding::percent_decode_str;
 use secrecy::{ExposeSecret, SecretString};
 use serde_json::Value;
 
-use super::Pl8;
+use super::{Pl8, UserSession};
 
 #[derive(Debug)]
 pub struct LoginError(pub String);
@@ -145,7 +146,9 @@ impl LoginFlow {
             return Err(err(e));
         }
         let session = self.session.take().ok_or_else(|| err("login failed, try again"))?;
-        let root = pl8.root(Some(&session)).await.ok().flatten();
+        let mut user_session = UserSession::new(session);
+        let root = pl8.root(Some(&mut user_session)).await.ok().flatten();
+        let session = user_session.into_cookie();
         let user = root.as_ref().and_then(|r| r.get("currentUser")).filter(|u| u.is_object());
         let Some(user) = user else {
             // Not logged in after all; nothing to end at pl8list.

backend/src/pl8/mod.rs

@@ -6,7 +6,7 @@
 //!
 //! 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.
+//! for login and for syncing with their account (feed, library, saves).
 
 pub mod login;
 pub mod normalize;
@@ -71,20 +71,17 @@ impl Pl8 {
     }
 
     pub fn request(&self, url: Url, session: Option<&SecretString>) -> reqwest::RequestBuilder {
-        let req = self.http.get(url);
-        match session {
-            Some(s) => req.header("Cookie", format!("_session={}", s.expose_secret())),
-            None => req,
-        }
+        with_session(self.http.get(url), session)
     }
 
-    /// JSON GET with retries on overload. 403/404 give `None`.
+    /// JSON GET with retries on overload. 403/404 give `None`. With a user session,
+    /// a redirect to the login page gives `SessionExpired`.
     pub async fn get_json(
         &self,
         segments: &[&str],
         params: &[(&str, String)],
         route: Option<&str>,
-        session: Option<&SecretString>,
+        mut session: Option<&mut UserSession>,
     ) -> anyhow::Result<Option<Value>> {
         let mut url = self.url(segments);
         {
@@ -98,7 +95,10 @@ impl Pl8 {
         }
         for attempt in 0..4u32 {
             self.throttle().await;
-            let res = self.request(url.clone(), session).send().await?;
+            let res = self.request(url.clone(), session.as_deref().map(|s| &s.cookie)).send().await?;
+            if let Some(s) = session.as_deref_mut() {
+                s.update(&res)?;
+            }
             match res.status() {
                 s if s.is_success() => {
                     let body = res.bytes().await?;
@@ -122,7 +122,7 @@ impl Pl8 {
     // Paginated loaders ignore `page` unless `pagination=true` is also set,
     // which is what the site's own useInfiniteLoading hook sends.
 
-    pub async fn root(&self, session: Option<&SecretString>) -> anyhow::Result<Option<Value>> {
+    pub async fn root(&self, session: Option<&mut UserSession>) -> anyhow::Result<Option<Value>> {
         self.get_json(&[], &[], Some("root"), session).await
     }
 
@@ -130,11 +130,23 @@ impl Pl8 {
         self.get_json(&[], &[], Some("routes/_index"), None).await
     }
 
-    pub async fn feed(&self, session: &SecretString, page: i64) -> anyhow::Result<Option<Value>> {
+    pub async fn feed(&self, session: &mut UserSession, page: i64) -> anyhow::Result<Option<Value>> {
         let p = [("page", page.to_string()), ("pagination", "true".into())];
         self.get_json(&["api", "feed"], &p, None, Some(session)).await
     }
 
+    /// Tracks the user saved on pl8list (`items`, `totalPages`).
+    pub async fn library_tracks(&self, session: &mut UserSession, page: i64) -> anyhow::Result<Option<Value>> {
+        let p = [("page", page.to_string()), ("pagination", "true".into())];
+        self.get_json(&["library", "tracks"], &p, Some("routes/library.tracks"), Some(session)).await
+    }
+
+    /// Artists the user saved (follows) on pl8list (`items`, `totalPages`).
+    pub async fn library_artists(&self, session: &mut UserSession, page: i64) -> anyhow::Result<Option<Value>> {
+        let p = [("page", page.to_string()), ("pagination", "true".into())];
+        self.get_json(&["library", "artists"], &p, Some("routes/library.artists"), Some(session)).await
+    }
+
     pub async fn artist(&self, handle: &str, page: i64) -> anyhow::Result<Option<Value>> {
         let p = [("page", page.to_string()), ("pagination", "true".into())];
         self.get_json(&[handle], &p, Some("routes/$handle"), None).await
@@ -157,6 +169,25 @@ impl Pl8 {
         self.get_json(&["search", "users"], &p, Some("routes/search.users"), None).await
     }
 
+    /// Save a track (`[handle, slug]`) or an artist (`[handle]`) to the user's pl8list
+    /// library. `addOnly` makes the toggle idempotent: it never removes anything.
+    pub async fn save(&self, session: &mut UserSession, target: &[&str]) -> anyhow::Result<()> {
+        let mut segments = vec!["api"];
+        segments.extend_from_slice(target);
+        segments.push("add-to-library");
+        let mut url = self.url(&segments);
+        url.query_pairs_mut().append_pair("addOnly", "true");
+        self.throttle().await;
+        let res = with_session(self.http.post(url), Some(&session.cookie)).send().await?;
+        session.update(&res)?;
+        let status = res.status();
+        let body: Value = res.json().await.unwrap_or(Value::Null);
+        if !status.is_success() || body.get("success").and_then(Value::as_bool) == Some(false) {
+            anyhow::bail!("pl8list refused the save (HTTP {status})");
+        }
+        Ok(())
+    }
+
     /// End a session at pl8list. Errors are ignored: we drop the session either way.
     pub async fn logout(&self, session: &SecretString) {
         let mut url = self.url(&["logout"]);
@@ -170,3 +201,64 @@ impl Pl8 {
         &self.http
     }
 }
+
+fn with_session(req: reqwest::RequestBuilder, session: Option<&SecretString>) -> reqwest::RequestBuilder {
+    match session {
+        Some(s) => req.header("Cookie", format!("_session={}", s.expose_secret())),
+        None => req,
+    }
+}
+
+/// pl8list no longer accepts the session; the user has to sign in again.
+#[derive(Debug)]
+pub struct SessionExpired;
+
+impl std::fmt::Display for SessionExpired {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        f.write_str("pl8list session expired")
+    }
+}
+
+impl std::error::Error for SessionExpired {}
+
+/// A signed-in pl8list session used on a user's behalf. pl8list may send a new
+/// cookie with any response; `changed` tells the owner to store it again.
+pub struct UserSession {
+    cookie: SecretString,
+    pub changed: bool,
+}
+
+impl UserSession {
+    pub fn new(cookie: SecretString) -> Self {
+        UserSession { cookie, changed: false }
+    }
+
+    pub fn cookie(&self) -> &SecretString {
+        &self.cookie
+    }
+
+    pub fn into_cookie(self) -> SecretString {
+        self.cookie
+    }
+
+    /// Take a refreshed cookie from `res`, and notice when pl8list sends us to its login.
+    fn update(&mut self, res: &reqwest::Response) -> Result<(), SessionExpired> {
+        let to_login = |p: &str| p == "/login" || p.starts_with("/login?");
+        let remix_redirect = res.headers().get("x-remix-redirect").and_then(|v| v.to_str().ok());
+        if to_login(res.url().path()) || remix_redirect.is_some_and(to_login) {
+            return Err(SessionExpired);
+        }
+        for c in res.headers().get_all("set-cookie") {
+            let Some(v) = c.to_str().ok().and_then(|c| c.strip_prefix("_session=")) else { continue };
+            let v = v.split(';').next().unwrap_or_default();
+            if v.is_empty() {
+                return Err(SessionExpired);
+            }
+            if v != self.cookie.expose_secret() {
+                self.cookie = SecretString::from(v.to_string());
+                self.changed = true;
+            }
+        }
+        Ok(())
+    }
+}

backend/src/ranker.rs

@@ -5,6 +5,7 @@
 //!     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)
+//!                  and whether you follow them
 //!     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
@@ -79,6 +80,8 @@ fn event_signal(kind: &str) -> f64 {
         _ => 0.0,
     }
 }
+/// Following an artist (here or on pl8list) counts like a few likes of their tracks.
+const FOLLOW_SIGNAL: f64 = 4.0;
 const REPEAT_ARTIST_PENALTY: f64 = 1.5;
 const RECENT_ARTIST_WINDOW: i64 = 6;
 
@@ -136,6 +139,9 @@ pub fn taste(conn: &Connection, user_id: i64) -> rusqlite::Result<Taste> {
             _ => {}
         }
     }
+    for handle in crate::db::follows(conn, user_id)? {
+        *artist.entry(handle).or_default() += FOLLOW_SIGNAL;
+    }
     let liked_set: HashSet<i64> = liked.iter().map(|l| l.0).collect();
     let positive: Vec<_> = completed.iter().filter(|c| !liked_set.contains(&c.0)).chain(liked.iter()).collect();
     let bpms: Vec<f64> = positive.iter().filter_map(|p| p.1).collect();

backend/src/sync.rs

@@ -0,0 +1,246 @@
+//! Sync with a user's pl8list account.
+//!
+//! In: the personal feed (tracks of artists they follow), saved tracks (become likes
+//! here) and saved artists (become follows). Out, only for devices that stay
+//! connected: likes and follows made here are saved to their pl8list library.
+//! Saving is add-only; nothing here ever removes anything from a pl8list account.
+//!
+//! Without "stay connected" a sign-in syncs once, then the session is logged out at
+//! pl8list. With it, the session is stored encrypted (`link.rs`) and used again only
+//! inside that device's own requests: a resync when the last one is older than
+//! `RESYNC_AFTER`, the resync button, and pushing likes and follows.
+
+use std::{
+    collections::HashSet,
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+
+use secrecy::SecretString;
+use serde_json::Value;
+
+use crate::{
+    crawler::Crawler,
+    db::{self, Db},
+    link::{LinkKey, Sealer},
+    pl8::{Pl8, SessionExpired, UserSession, normalize},
+};
+
+/// 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;
+const DEADLINE: Duration = Duration::from_secs(120);
+
+pub struct Syncer {
+    pl8: Arc<Pl8>,
+    db: Db,
+    crawler: Arc<Crawler>,
+    pub key: LinkKey,
+    /// Users with a sync running, so a burst of requests starts only one.
+    running: Mutex<HashSet<i64>>,
+}
+
+/// What to do with the pl8list session once a task is done with it.
+enum After {
+    Logout,
+    Store(Sealer),
+}
+
+fn text(v: &Value, k: &str) -> Option<String> {
+    v.get(k).and_then(Value::as_str).filter(|s| !s.is_empty()).map(String::from)
+}
+
+/// Library items are `{track, ...}` like feed items; accept a bare track too.
+fn library_track(item: &Value) -> Option<db::Track> {
+    normalize::item(item).or_else(|| normalize::item(&serde_json::json!({ "track": item })))
+}
+
+fn library_artist(item: &Value) -> Option<String> {
+    text(item, "handle")
+        .or_else(|| item.get("user").and_then(|u| text(u, "handle")))
+        .or_else(|| item.get("artist").and_then(|u| text(u, "handle")))
+}
+
+impl Syncer {
+    pub fn new(pl8: Arc<Pl8>, db: Db, crawler: Arc<Crawler>, key: LinkKey) -> Arc<Self> {
+        Arc::new(Syncer { pl8, db, crawler, key, running: Mutex::new(HashSet::new()) })
+    }
+
+    fn claim(&self, user_id: i64) -> bool {
+        self.running.lock().unwrap_or_else(|e| e.into_inner()).insert(user_id)
+    }
+
+    fn release(&self, user_id: i64) {
+        self.running.lock().unwrap_or_else(|e| e.into_inner()).remove(&user_id);
+    }
+
+    /// Sync once with a fresh sign-in session, then log it out at pl8list.
+    pub fn once(self: &Arc<Self>, user_id: i64, session: SecretString) {
+        self.spawn_sync(user_id, session, After::Logout);
+    }
+
+    /// Sync with a connected device's session. Returns false if one is already running.
+    pub fn linked(self: &Arc<Self>, sealer: Sealer, session: SecretString) -> bool {
+        self.spawn_sync(sealer.user_id, session, After::Store(sealer))
+    }
+
+    fn spawn_sync(self: &Arc<Self>, user_id: i64, session: SecretString, after: After) -> bool {
+        if !self.claim(user_id) {
+            if let After::Logout = after {
+                // Never keep a one-time session around just because a sync is running.
+                let pl8 = self.pl8.clone();
+                tokio::spawn(async move { pl8.logout(&session).await });
+            }
+            return false;
+        }
+        let me = self.clone();
+        tokio::spawn(async move {
+            if let After::Store(sealer) = &after {
+                let hash = sealer.token_hash.clone();
+                if let Err(e) = me.db.call(move |c| db::mark_synced(c, &hash)).await {
+                    tracing::warn!("marking sync: {e:#}");
+                }
+            }
+            let mut s = UserSession::new(session);
+            let res = tokio::time::timeout(DEADLINE, me.read_account(user_id, &mut s)).await;
+            let expired = match &res {
+                Ok(Ok((feed, likes, follows))) => {
+                    tracing::info!("synced pl8list: {feed} feed pages, {likes} new likes, {follows} new follows");
+                    false
+                }
+                Ok(Err(e)) => {
+                    tracing::warn!("pl8list sync failed: {e:#}");
+                    e.is::<SessionExpired>()
+                }
+                Err(_) => {
+                    tracing::warn!("pl8list sync timed out");
+                    false
+                }
+            };
+            me.finish(s, after, expired).await;
+            me.release(user_id);
+        });
+        true
+    }
+
+    /// Save a liked track to the user's pl8list library.
+    pub fn push_like(self: &Arc<Self>, sealer: Sealer, session: SecretString, handle: String, slug: String) {
+        let me = self.clone();
+        tokio::spawn(async move {
+            let mut s = UserSession::new(session);
+            let res = me.pl8.save(&mut s, &[&handle, &slug]).await;
+            me.after_push(s, sealer, res).await;
+        });
+    }
+
+    /// Save a followed artist to the user's pl8list library.
+    pub fn push_follow(self: &Arc<Self>, sealer: Sealer, session: SecretString, handle: String) {
+        let me = self.clone();
+        tokio::spawn(async move {
+            let mut s = UserSession::new(session);
+            let res = me.pl8.save(&mut s, &[&handle]).await;
+            me.after_push(s, sealer, res).await;
+        });
+    }
+
+    async fn after_push(&self, s: UserSession, sealer: Sealer, res: anyhow::Result<()>) {
+        let expired = match &res {
+            Ok(()) => false,
+            Err(e) => {
+                tracing::warn!("pl8list save failed: {e:#}");
+                e.is::<SessionExpired>()
+            }
+        };
+        self.finish(s, After::Store(sealer), expired).await;
+    }
+
+    /// Log a one-time session out, or store a refreshed cookie, or forget an expired one.
+    async fn finish(&self, s: UserSession, after: After, expired: bool) {
+        match after {
+            After::Logout => {
+                if !expired {
+                    self.pl8.logout(s.cookie()).await;
+                }
+            }
+            After::Store(sealer) => {
+                let hash = sealer.token_hash.clone();
+                let res = if expired {
+                    tracing::info!("pl8list session expired, device disconnected");
+                    self.db.call(move |c| db::clear_link(c, &hash)).await
+                } else if s.changed {
+                    let blob = sealer.seal(s.cookie());
+                    self.db.call(move |c| db::set_link(c, &hash, &blob, None)).await
+                } else {
+                    Ok(())
+                };
+                if let Err(e) = res {
+                    tracing::warn!("storing pl8list session: {e:#}");
+                }
+            }
+        }
+    }
+
+    /// 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;
+        for page in 1..=MAX_PAGES {
+            let d = self.pl8.feed(s, page).await?.unwrap_or(Value::Null);
+            let rows = normalize::items(d.get("tracks"));
+            self.crawler.discovered_artists(rows.iter().map(|r| r.handle.clone()));
+            self.db.call(move |c| db::upsert_tracks(c, &rows, None)).await?;
+            feed_pages = page;
+            if page >= d.get("totalPages").and_then(Value::as_i64).unwrap_or(0) {
+                break;
+            }
+        }
+
+        let mut likes = 0;
+        for page in 1..=MAX_PAGES {
+            let d = self.pl8.library_tracks(s, page).await?.unwrap_or(Value::Null);
+            let rows: Vec<db::Track> =
+                d.get("items").and_then(Value::as_array).map(|a| a.iter().filter_map(library_track).collect()).unwrap_or_default();
+            let ids: Vec<i64> = rows.iter().map(|r| r.id).collect();
+            self.crawler.discovered_artists(rows.iter().map(|r| r.handle.clone()));
+            likes += self
+                .db
+                .call(move |c| {
+                    db::upsert_tracks(c, &rows, None)?;
+                    db::import_likes(c, user_id, &ids)
+                })
+                .await?;
+            if page >= d.get("totalPages").and_then(Value::as_i64).unwrap_or(0) {
+                break;
+            }
+        }
+
+        let mut follows = 0;
+        for page in 1..=MAX_PAGES {
+            let d = self.pl8.library_artists(s, page).await?.unwrap_or(Value::Null);
+            let handles: Vec<String> =
+                d.get("items").and_then(Value::as_array).map(|a| a.iter().filter_map(library_artist).collect()).unwrap_or_default();
+            self.crawler.discovered_artists(handles.clone());
+            follows += self.db.call(move |c| db::add_follows(c, user_id, &handles, "pl8list")).await?;
+            if page >= d.get("totalPages").and_then(Value::as_i64).unwrap_or(0) {
+                break;
+            }
+        }
+        Ok((feed_pages, likes, follows))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn library_items_in_either_shape() {
+        let wrapped = json!({"track": {"id": 1, "slug": "s", "user": {"handle": "h"}}});
+        let bare = json!({"id": 2, "slug": "s", "user": {"handle": "h"}});
+        assert_eq!(library_track(&wrapped).unwrap().id, 1);
+        assert_eq!(library_track(&bare).unwrap().id, 2);
+        assert_eq!(library_artist(&json!({"handle": "a"})).as_deref(), Some("a"));
+        assert_eq!(library_artist(&json!({"user": {"handle": "b"}})).as_deref(), Some("b"));
+        assert_eq!(library_artist(&json!({"name": "c"})), None);
+    }
+}