diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 9f0cc51..5b658d7 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A page of items asks for its count beside the items, and the feed list asks its five questions at once, where each waited for the one before (#346). - A Signal's count of items and of unread ones are worked out in one pass, not two (#348). - Artwork kept on disk is read and written off the threads that answer requests, and two requests for the same missing artwork fetch it once (#345, #349). +- Unread lists and counts read a feed's items from each person's oldest unread one on, which the daily cleanup works out, not the whole back catalogue (#347). ### Fixed diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index 66de3b3..ee4df67 100644 --- a/crates/ipx/src/db.rs +++ b/crates/ipx/src/db.rs @@ -86,6 +86,8 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> { // the expression it is asked by. Without it, each wake read all of entries, 605 MB, 566 ms // to find nothing (#274). "CREATE INDEX IF NOT EXISTS entries_recent ON entries (feed_id, (coalesce(published, first_seen)))", + // Unread lists read a feed's items from `subscriptions.unread_from` on (#347). + "CREATE INDEX IF NOT EXISTS entries_seen ON entries (feed_id, first_seen)", ] { orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?; } @@ -166,6 +168,10 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> { let sql = "ALTER TABLE signals ADD COLUMN icon TEXT"; orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?; } + if orm.execute_unprepared("SELECT unread_from FROM subscriptions LIMIT 0").await.is_err() { + let sql = "ALTER TABLE subscriptions ADD COLUMN unread_from BIGINT"; + orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?; + } if orm.execute_unprepared("SELECT pause_ms FROM deadair_prefs LIMIT 0").await.is_err() { let sql = "ALTER TABLE deadair_prefs ADD COLUMN pause_ms BIGINT"; orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?; @@ -541,6 +547,7 @@ impl Db { "unread", "SELECT e.feed_id, count(*) AS n FROM entries e JOIN subscriptions sub ON sub.user_id = $1 AND sub.feed_id = e.feed_id + AND e.first_seen >= coalesce(sub.unread_from, 0) LEFT JOIN entry_state s ON s.user_id = $1 AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE NOT coalesce(s.read, false) @@ -1236,8 +1243,20 @@ fn entries_from( } None => "true".into(), }; + // Unread starts from where this person's unread begins in each feed, so it reads the few + // items since, not every one: 11,727 read for 44 unread, 32 ms (#347). Every list is of + // subscribed feeds, so the join takes nothing away. + let since = if filter == Filter::Unread { + format!( + "JOIN subscriptions w ON w.user_id = {user} AND w.feed_id = e.feed_id + AND e.first_seen >= coalesce(w.unread_from, 0)" + ) + } else { + String::new() + }; format!( "FROM entries e + {since} LEFT JOIN entry_state s ON s.user_id = {user} AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE {scope} AND {} AND {search} AND {}", filter.sql(), @@ -2278,9 +2297,37 @@ impl Db { vec![user_id.into(), feed_id.into(), guid.into(), on.into()], ) .await?; + // An item marked unread again may be older than where this person's unread was known to + // start, so that is forgotten until the next reap works it out (`advance_unread_from`). + if matches!(field, EntryFlag::Read) && !on { + self.exec( + "UPDATE subscriptions SET unread_from = NULL + WHERE user_id = $1 AND feed_id = $2 AND unread_from IS NOT NULL", + vec![user_id.into(), feed_id.into()], + ) + .await?; + } Ok(()) } + /// Moves each subscription's `unread_from` up to its oldest unread item, hidden ones included, + /// since a block list can be lifted. No more than an hour ago: an item a scan is storing now, + /// not yet visible to this statement, is first seen about now. Returns how many moved. + pub async fn advance_unread_from(&self) -> Result { + let cutoff = now() - 3600; + self.exec( + "UPDATE subscriptions SET unread_from = ( + SELECT CASE WHEN min(e.first_seen) < $1 THEN min(e.first_seen) ELSE $1 END + FROM entries e + LEFT JOIN entry_state s + ON s.user_id = subscriptions.user_id AND s.feed_id = e.feed_id AND s.guid = e.guid + WHERE e.feed_id = subscriptions.feed_id AND NOT coalesce(s.read, false)) + WHERE true", + vec![cutoff.into()], + ) + .await + } + /// How many files this feed has on disk. Decides whether a feed dropped from an OPML /// can be removed or must be kept. pub async fn downloaded_count(&self, feed_id: &str) -> Result { @@ -3019,6 +3066,51 @@ pub fn now() -> i64 { mod tests { use super::*; + #[tokio::test] + async fn unread_starts_where_it_was_worked_out_and_loses_nothing() { + let db = Db::memory().await.unwrap(); + let me = db.create_user("me", None, false).await.unwrap(); + db.exec_for_test( + "INSERT INTO feeds (id, url) VALUES ('f', 'http://x/f.xml'); + INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','old',100),('f','mid',200),('f','new',300);", + ) + .await + .unwrap(); + db.subscribe(me, "f").await.unwrap(); + db.set_entry_flag(me, "f", "old", EntryFlag::Read, true).await.unwrap(); + let unread = async || { + let order = order_sql("published", "desc"); + let rows = db.entries_of(me, None, Filter::Unread, None, 0, 50, &order).await.unwrap(); + let mut guids: Vec = rows.into_iter().map(|r| r.guid).collect(); + guids.sort(); + let count = db.count_of(me, None, Filter::Unread, None).await.unwrap(); + let listed = db.feed_list(me, None).await.unwrap()["f"].unread; + assert_eq!(count, guids.len() as i64); + assert_eq!(listed, count, "the feed list counts what the list shows"); + guids + }; + assert_eq!(unread().await, ["mid", "new"]); + + db.advance_unread_from().await.unwrap(); + let from = async || -> Option { + db.rows("SELECT unread_from FROM subscriptions", vec![]).await.unwrap()[0] + .try_get("", "unread_from") + .unwrap() + }; + assert_eq!(from().await, Some(200), "the oldest unread"); + assert_eq!(unread().await, ["mid", "new"], "nothing unread goes missing (#347)"); + + // An old item marked unread again comes back, though it is older than where unread began. + db.set_entry_flag(me, "f", "old", EntryFlag::Read, false).await.unwrap(); + assert_eq!(unread().await, ["mid", "new", "old"]); + + // With everything read, it starts no later than an hour ago. + db.mark_all_read(me, &["f".into()]).await.unwrap(); + db.advance_unread_from().await.unwrap(); + assert!(from().await.is_some_and(|t| t <= now() - 3600)); + assert!(unread().await.is_empty()); + } + #[tokio::test] async fn a_new_password_signs_the_account_out_everywhere_but_keeps_its_tokens() { let db = Db::memory().await.unwrap(); @@ -3620,6 +3712,8 @@ mod tests { (1,'f','b','u1','/tmp/b','done'); -- Read and starred belong to a person now, so say which one. INSERT INTO users (id, name, is_admin) VALUES (7,'reader',true); + -- Every list is of feeds its reader subscribes to; Unread joins the subscription. + INSERT INTO subscriptions (user_id, feed_id) VALUES (7,'f'); INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES (7,'f','b',true,false,0), (7,'f','c',true,true,0), diff --git a/crates/ipx/src/entity.rs b/crates/ipx/src/entity.rs index dfd6410..4b4a6cc 100644 --- a/crates/ipx/src/entity.rs +++ b/crates/ipx/src/entity.rs @@ -230,6 +230,10 @@ pub mod subscriptions { /// Left out of every Signal, and never put to Jev for them, as `hide_from_all` (#289). #[sea_orm(default_value = false)] pub hide_from_signals: bool, + /// Every item of the feed first seen before this is read by this person (#347): unread + /// lists start here instead of reading the whole back catalogue. NULL knows nothing, and + /// the daily reap moves it up (`Db::advance_unread_from`). + pub unread_from: Option, } owned_by_user!(); diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index 6c6ec22..6b3689a 100644 --- a/crates/ipx/src/main.rs +++ b/crates/ipx/src/main.rs @@ -651,6 +651,11 @@ async fn daemon( ); } + // Unread lists start from here, and a reap may be a day off (#347). Slower without, never wrong. + if let Err(e) = ctx.db.advance_unread_from().await { + tracing::warn!(error = %format!("{e:#}"), "could not work out where unread items start"); + } + if let Some(admin) = ctx.db.users().await?.into_iter().find(|u| u.is_admin) { let catalogue: Vec = ctx.cfg().feeds.keys().cloned().collect(); match ctx.db.adopt_catalogue(admin.id, &catalogue).await { @@ -1238,6 +1243,9 @@ async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { span.record("entries", r.entries_pruned); if !dry_run { tracing::Instrument::instrument(clean_directory(ctx), db::step("clean_directory")).await?; + let moved = + tracing::Instrument::instrument(ctx.db.advance_unread_from(), db::step("advance_unread_from")).await?; + tracing::debug!(moved, "unread lists start from each person's oldest unread item"); // Off the async threads: it walks every file kept, on FUSE (#345). let (span, limit) = (db::step("trim_artwork"), ctx.cfg().general.art_cache_mb * 1_048_576); let gone = tokio::task::spawn_blocking(move || span.in_scope(|| art::trim(&art::dir(), limit))).await?;