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/<key>) 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 <noreply@anthropic.com>

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
 

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<String>) -> 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<String> = ["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);

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<WebState>,
     user: crate::db::User,
 ) -> Sse<impl futures_util::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
     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::time::Instant>, std::collections::HashSet<String>) = (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 {
+    let stream =
-        loop {
+        futures_util::stream::unfold((rx, state, user, looked), |(mut rx, state, mut user, mut looked)| async move {
-            match rx.recv().await {
+            loop {
-                Ok(ev) => {
+                match rx.recv().await {
-                    let Ok(data) = serde_json::to_string(&ev) else { continue };
+                    Ok(ev) => {
-                    let mut out = vec![Ok::<_, std::convert::Infallible>(SseEvent::default().data(data))];
+                        if looked.0.is_none_or(|at| at.elapsed() >= EVENTS_RECHECK) {
-                    // A feed that changed goes out as this person's row for it, which the page
+                            user = state.ctx.db.user_by_id(user.id).await.ok().flatten()?;
-                    // puts in place of the old one: it reloaded the whole list after each.
+                            let subs = state.ctx.db.subscriptions_for(user.id).await.unwrap_or_default();
-                    if let Some(feed) = changed_feed(&ev)
+                            looked = (Some(std::time::Instant::now()), subs.into_iter().map(|s| s.feed_id).collect());
-                        && let Ok(rows) = feed_rows(&state, &user, Some(feed)).await
+                        }
-                        && let Some(row) = rows.first()
+                        if !ev.is_for(&user, &looked.1) {
-                        && let Ok(data) = serde_json::to_string(&serde_json::json!({ "ev": "feed_row", "row": row }))
+                            continue;
-                    {
+                        }
-                        out.push(Ok(SseEvent::default().data(data)));
+                        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())
 }