From 91de1114d8ddab5e0f6f6d89235ed554d566af94 Mon Sep 17 00:00:00 2001 From: rays Date: Fri, 23 Oct 2026 14:51:12 +0000 Subject: [PATCH] Live updates go to those who read the feed, while their account exists (#325 #335) /api/events sent every daemon event to every open page. Event::redacted strips the query and the password from addresses, not the path, so a paid feed's key in its path (feeds.supercast.com/feeds/) reached everyone signed in through its FeedError or FeedMoved (#325). And the stream checked who was asking once, when it opened: a removed account's page kept getting events until it closed (#335). Each stream now looks its account and subscriptions up again at most every 30 s, as an event arrives, ends when the account is gone, and sends an event about a feed only to someone subscribed to it. Admins get everything, for the log; Reaped, a file path, goes to admins only; SignalHits to the person it is for. Co-Authored-By: Claude Opus 5.5 --- crates/ipx/CHANGELOG.md | 2 ++ crates/ipx/src/ipc.rs | 57 +++++++++++++++++++++++++++++++++++++++ crates/ipx/src/web.rs | 59 ++++++++++++++++++++++++++--------------- 3 files changed, 96 insertions(+), 22 deletions(-) diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 3bbef01..b6bef30 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -35,6 +35,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - An OPML or Patreon list may bring in up to 2,000 feeds; one with more shows an error and adds none. A list of tens of thousands held up checking for everyone, and adding its feeds read the whole feeds table once for each (#337). - A request that changes something is refused when a browser sent it from another site. A page on another site, a sibling subdomain included, could mark everything read, start downloads, subscribe you to a Directory feed or sign you out (#336). - `ipx user passwd` signs the account out of every browser, and says how many of its API tokens still work. A session signed in with the old password went on working (#334). +- Live updates about a feed go only to the people who read it, and admins. Everyone signed in had every feed's, errors included, which carried a paid feed's address with its key (#325). +- An open page stops getting live updates within half a minute of its account being removed. They went on until the page was closed (#335). ## [1.4.0] - 2026-10-19 diff --git a/crates/ipx/src/ipc.rs b/crates/ipx/src/ipc.rs index ada7983..5824824 100644 --- a/crates/ipx/src/ipc.rs +++ b/crates/ipx/src/ipc.rs @@ -104,6 +104,38 @@ pub enum Event { } impl Event { + /// The feed this is about, for sending it only to those who read that feed. + pub fn feed(&self) -> Option<&str> { + match self { + Event::FeedStart { feed } + | Event::FeedSkip { feed, .. } + | Event::FeedDone { feed, .. } + | Event::FeedError { feed, .. } + | Event::FeedMoved { feed, .. } + | Event::Progress { feed, .. } + | Event::DownloadDone { feed, .. } + | Event::DownloadError { feed, .. } + | Event::TorrentDeferred { feed, .. } + | Event::DeadairDone { feed, .. } => Some(feed), + _ => None, + } + } + + /// Whether a page signed in as this person is sent it. Everyone had every event, so a key in + /// a paid feed's path, which `redacted` keeps, reached people who do not read the feed (#325). + /// An admin has them all, for the log. + pub fn is_for(&self, user: &crate::db::User, mine: &std::collections::HashSet) -> bool { + if user.is_admin { + return true; + } + match self { + Event::SignalHits { user: u, .. } => *u == user.id, + // Paths of files on disk, everyone's. + Event::Reaped { .. } => false, + ev => ev.feed().is_none_or(|f| mine.contains(f)), + } + } + pub fn is_terminal(&self) -> bool { matches!(self, Event::ScanDone { .. } | Event::ReapDone { .. } | Event::Status { .. }) } @@ -458,6 +490,31 @@ pub async fn proxy(path: &Path, cmd: &Command) -> Result<()> { mod tests { use super::*; + #[test] + fn a_feeds_events_go_to_those_who_read_it() { + let person = |id, is_admin| crate::db::User { + id, + name: "x".into(), + pass_hash: None, + is_admin, + created: None, + last_login: None, + }; + let (me, admin) = (person(1, false), person(2, true)); + let mine: std::collections::HashSet = ["mine".to_string()].into(); + let error = |feed: &str| Event::FeedError { + feed: feed.into(), + msg: "error sending request for url (https://feeds.supercast.com/feeds/KEY)".into(), + }; + assert!(error("mine").is_for(&me, &mine)); + assert!(!error("theirs").is_for(&me, &mine), "a paid feed's key in its path (#325)"); + assert!(error("theirs").is_for(&admin, &mine), "an admin's log has everything"); + assert!(Event::ScanDone { feeds: 3, only: None, forced: false }.is_for(&me, &mine)); + assert!(Event::SignalHits { user: 1, hits: 2 }.is_for(&me, &mine)); + assert!(!Event::SignalHits { user: 3, hits: 2 }.is_for(&me, &mine)); + assert!(!Event::Reaped { path: "/downloads/Theirs/ep.mp3".into(), bytes: 1 }.is_for(&me, &mine)); + } + #[test] fn a_feed_not_due_stays_off_the_live_updates() { let (tx, mut rx) = broadcast::channel(4); diff --git a/crates/ipx/src/web.rs b/crates/ipx/src/web.rs index e09c958..31d6a62 100644 --- a/crates/ipx/src/web.rs +++ b/crates/ipx/src/web.rs @@ -2018,37 +2018,52 @@ async fn fetch_now( } /// The same broadcast the socket clients read, as server-sent events. +const EVENTS_RECHECK: std::time::Duration = std::time::Duration::from_secs(30); + async fn events( State(state): State, user: crate::db::User, ) -> Sse>> { let rx = state.events.subscribe(); + // Who this is, and what they read, looked up again every half minute: the stream outlives + // the request that opened it, and went on after the account was removed (#335). + let looked: (Option, std::collections::HashSet) = (None, Default::default()); // A client that falls behind skips what it missed rather than being cut off. - let stream = futures_util::stream::unfold((rx, state, user), |(mut rx, state, user)| async move { - loop { - match rx.recv().await { - Ok(ev) => { - let Ok(data) = serde_json::to_string(&ev) else { continue }; - let mut out = vec![Ok::<_, std::convert::Infallible>(SseEvent::default().data(data))]; - // A feed that changed goes out as this person's row for it, which the page - // puts in place of the old one: it reloaded the whole list after each. - if let Some(feed) = changed_feed(&ev) - && let Ok(rows) = feed_rows(&state, &user, Some(feed)).await - && let Some(row) = rows.first() - && let Ok(data) = serde_json::to_string(&serde_json::json!({ "ev": "feed_row", "row": row })) - { - out.push(Ok(SseEvent::default().data(data))); + let stream = + futures_util::stream::unfold((rx, state, user, looked), |(mut rx, state, mut user, mut looked)| async move { + loop { + match rx.recv().await { + Ok(ev) => { + if looked.0.is_none_or(|at| at.elapsed() >= EVENTS_RECHECK) { + user = state.ctx.db.user_by_id(user.id).await.ok().flatten()?; + let subs = state.ctx.db.subscriptions_for(user.id).await.unwrap_or_default(); + looked = (Some(std::time::Instant::now()), subs.into_iter().map(|s| s.feed_id).collect()); + } + if !ev.is_for(&user, &looked.1) { + continue; + } + let Ok(data) = serde_json::to_string(&ev) else { continue }; + let mut out = vec![Ok::<_, std::convert::Infallible>(SseEvent::default().data(data))]; + // A feed that changed goes out as this person's row for it, which the page + // puts in place of the old one: it reloaded the whole list after each. + if let Some(feed) = changed_feed(&ev) + && let Ok(rows) = feed_rows(&state, &user, Some(feed)).await + && let Some(row) = rows.first() + && let Ok(data) = + serde_json::to_string(&serde_json::json!({ "ev": "feed_row", "row": row })) + { + out.push(Ok(SseEvent::default().data(data))); + } + return Some((futures_util::stream::iter(out), (rx, state, user, looked))); } - return Some((futures_util::stream::iter(out), (rx, state, user))); + // Counted for /metrics: a page that misses updates shows stale counts (#195). + Err(broadcast::error::RecvError::Lagged(n)) => { + crate::metrics::STATS.events_missed.fetch_add(n, std::sync::atomic::Ordering::Relaxed); + } + Err(broadcast::error::RecvError::Closed) => return None, } - // Counted for /metrics: a page that misses updates shows stale counts (#195). - Err(broadcast::error::RecvError::Lagged(n)) => { - crate::metrics::STATS.events_missed.fetch_add(n, std::sync::atomic::Ordering::Relaxed); - } - Err(broadcast::error::RecvError::Closed) => return None, } - } - }); + }); Sse::new(futures_util::StreamExt::flatten(stream)).keep_alive(axum::response::sse::KeepAlive::default()) }