From 4bd9c1e0d2b523d9348da1aea97e0f6aa276d1bf Mon Sep 17 00:00:00 2001 From: hART <875f39cab0eec1bd512cf98f99ba5d5dc0863e090ac893cdaee499c2f1c25d03@buzz.block.builderlab.xyz> Date: Tue, 6 Oct 2026 21:13:28 -0700 Subject: [PATCH] prototype: sidebar reports unread presence, not counts Booleans for ordinary unread and attention, existence probes instead of counts to a cap, mentions from event_mentions, threads pruned by root last_reply_received_at. Benchmark scaffold included; not for merge. --- crates/buzz-db/src/store/event.rs | 27 +-- .../personal_read/arrival_postgres_tests.rs | 26 +-- .../personal_read/bench_postgres_tests.rs | 199 ++++++++++++++++++ .../src/store/personal_read/classification.rs | 10 +- .../src/store/personal_read/context.rs | 7 +- crates/buzz-db/src/store/personal_read/mod.rs | 3 + .../buzz-db/src/store/personal_read/model.rs | 27 +-- .../src/store/personal_read/postgres_tests.rs | 52 ++--- .../src/store/personal_read/projection.rs | 163 +++++++------- .../projection_postgres_tests.rs | 32 ++- .../personal_read/threads_postgres_tests.rs | 59 +++--- .../src/api/buzz_v1/postgres_tests.rs | 2 +- crates/buzz-relay/src/nip11.rs | 5 +- migrations/0056_personal_read_state.sql | 3 + schema/schema.sql | 1 + 15 files changed, 416 insertions(+), 200 deletions(-) create mode 100644 crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index a6d7cb8516..c6d9ac1f24 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1696,20 +1696,21 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( .execute(&mut **tx) .await?; - if let Some(root_id) = meta.root_event_id { - sqlx::query( - r#" - UPDATE thread_metadata - SET descendant_count = descendant_count + 1 - WHERE community_id = $1 AND event_id = $2 - "#, - ) - .bind(community_id.as_uuid()) - .bind(root_id) - .execute(&mut **tx) - .await?; - } let root = meta.root_event_id.unwrap_or(pid); + sqlx::query( + r#" + UPDATE thread_metadata + SET descendant_count = descendant_count + $3::int, + last_reply_received_at = GREATEST(last_reply_received_at, $4) + WHERE community_id = $1 AND event_id = $2 + "#, + ) + .bind(community_id.as_uuid()) + .bind(root) + .bind(i32::from(meta.root_event_id.is_some())) + .bind(received_at) + .execute(&mut **tx) + .await?; crate::personal_read::record_reply( tx, community_id, diff --git a/crates/buzz-db/src/store/personal_read/arrival_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/arrival_postgres_tests.rs index d8b7e0dbce..ef3fd98905 100644 --- a/crates/buzz-db/src/store/personal_read/arrival_postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/arrival_postgres_tests.rs @@ -154,7 +154,7 @@ async fn late_arrival_with_old_author_time_is_unread() { states(&db, community, &actor, channel, &[&read, &late]).await, ["read", "unread"] ); - assert_eq!(sidebar(&db, community, &actor).await.unread, 1); + assert!(sidebar(&db, community, &actor).await.unread); } #[tokio::test] @@ -179,7 +179,7 @@ async fn future_dated_anchor_does_not_swallow_later_arrivals() { states(&db, community, &actor, channel, &[&earlier, &ahead, &later]).await, ["read", "read", "unread"] ); - assert_eq!(sidebar(&db, community, &actor).await.unread, 1); + assert!(sidebar(&db, community, &actor).await.unread); } /// Marking the sidebar's own latest message must clear the badge even when the @@ -192,7 +192,7 @@ async fn mark_as_read_with_the_sidebar_anchor_clears_a_late_arrival() { arrive(&pool, community, &first, now - 60).await; post(&db, &pool, community, channel, now + 1, now - 40).await; post(&db, &pool, community, channel, now - 600, now - 30).await; - assert_eq!(sidebar(&db, community, &actor).await.unread, 3); + assert!(sidebar(&db, community, &actor).await.unread); let anchor = sidebar(&db, community, &actor) .await @@ -200,7 +200,7 @@ async fn mark_as_read_with_the_sidebar_anchor_clears_a_late_arrival() { .expect("a channel with messages has a latest message"); apply(&db, community, &actor, mark_channel_read(channel, anchor)).await; - assert_eq!(sidebar(&db, community, &actor).await.unread, 0); + assert!(!sidebar(&db, community, &actor).await.unread); } #[tokio::test] @@ -248,7 +248,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { .await .unwrap(); let row = sidebar(&db, community, &actor).await; - assert_eq!((row.unread, row.unread_thread_count), (2, 1)); + assert_eq!((row.unread, row.attention, row.threads.len()), (false, true, 1)); assert_eq!(row.threads.len(), 1); let root_id = root.id.to_hex(); @@ -262,7 +262,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { .await; let row = sidebar(&db, community, &actor).await; - assert_eq!((row.unread, row.unread_thread_count), (0, 0)); + assert_eq!((row.unread, row.attention, row.threads.len()), (false, false, 0)); assert!(row.threads.is_empty()); } @@ -283,11 +283,11 @@ async fn mark_as_read_clears_a_late_arrival_behind_the_latest_probe() { .latest_message_id .unwrap(); apply(&db, community, &actor, mark_channel_read(channel, anchor)).await; - assert_eq!(sidebar(&db, community, &actor).await.unread, 0); + assert!(!sidebar(&db, community, &actor).await.unread); let late = post(&db, &pool, community, channel, now - 600, now - 10).await; let row = sidebar(&db, community, &actor).await; - assert_eq!(row.unread, 1); + assert!(row.unread); assert_eq!(row.latest_message_id, Some(late.id.to_hex())); apply( &db, @@ -296,7 +296,7 @@ async fn mark_as_read_clears_a_late_arrival_behind_the_latest_probe() { mark_channel_read(channel, late.id.to_hex()), ) .await; - assert_eq!(sidebar(&db, community, &actor).await.unread, 0); + assert!(!sidebar(&db, community, &actor).await.unread); } /// Reactions never fill the latest probe. @@ -321,7 +321,7 @@ async fn latest_message_behind_newer_reactions_is_found() { let row = sidebar(&db, community, &actor).await; assert_eq!(row.latest_message_id, Some(only.id.to_hex())); assert_eq!(row.latest_message_at, Some((now - 600) as i64)); - assert_eq!(row.unread, 1); + assert!(row.unread); } /// Store `event` as having arrived at exactly `seconds` plus `micros`. Built @@ -355,7 +355,7 @@ async fn arrive_exact( /// Two messages with the same author time arriving within one second; marks /// the one at `pick` and returns both channel states. -async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec, u32) { +async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec, bool) { let (db, pool, community, channel, actor, first) = fixture().await; let now = first.created_at.as_secs(); let second = post(&db, &pool, community, channel, now, now).await; @@ -383,7 +383,7 @@ async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec, async fn arrivals_one_microsecond_apart_in_the_same_second_are_ordered() { assert_eq!( mark_within_one_second([500_000, 500_001], 0).await, - (vec!["read".to_owned(), "unread".to_owned()], 1) + (vec!["read".to_owned(), "unread".to_owned()], true) ); } @@ -395,7 +395,7 @@ async fn marking_either_of_two_identical_arrivals_reads_both() { for pick in [0, 1] { assert_eq!( mark_within_one_second([500_000, 500_000], pick).await, - (vec!["read".to_owned(), "read".to_owned()], 0), + (vec!["read".to_owned(), "read".to_owned()], false), "marked index {pick}" ); } diff --git a/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs new file mode 100644 index 0000000000..dd34fb7697 --- /dev/null +++ b/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs @@ -0,0 +1,199 @@ +//! Side-by-side benchmark (not for merge). Identical file in both branches. +//! BENCH_THREADS (default 5000) followed threads, 3 replies each. +use super::postgres_tests::{apply, channel_read, mention}; +use super::*; +use buzz_core::CommunityId; +use crate::{ + channel::{ChannelType, ChannelVisibility}, + Db, +}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use sqlx::postgres::PgPoolOptions; +use std::time::{Duration, Instant}; +use uuid::Uuid; + +fn ts() -> u64 { + nostr::Timestamp::now().as_secs() +} + +fn ev(author: &Keys, at: u64, tags: Vec) -> nostr::Event { + EventBuilder::new(Kind::Custom(9), format!("bench {at} {}", Uuid::new_v4())) + .tags(tags) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(author) + .unwrap() +} + +async fn top(db: &Db, c: CommunityId, ch: Uuid, e: &nostr::Event) -> Duration { + let t = Instant::now(); + db.insert_event(c, e, Some(ch)).await.unwrap(); + t.elapsed() +} + +async fn rep(db: &Db, c: CommunityId, ch: Uuid, root: &nostr::Event, e: &nostr::Event) -> Duration { + let stamp = |e: &nostr::Event| chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap(); + let t = Instant::now(); + db.insert_event_with_thread_metadata( + c, + e, + Some(ch), + Some(crate::event::ThreadMetadataParams { + event_id: e.id.as_bytes(), + event_created_at: stamp(e), + channel_id: ch, + parent_event_id: Some(root.id.as_bytes()), + parent_event_created_at: Some(stamp(root)), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(stamp(root)), + depth: 1, + broadcast: false, + }), + ) + .await + .unwrap(); + t.elapsed() +} + +fn stats(label: &str, mut v: Vec) { + v.sort(); + let p = |q: f64| v[((v.len() as f64 - 1.0) * q) as usize].as_secs_f64() * 1000.0; + println!( + "BENCH {label}: n={} p50={:.3}ms p95={:.3}ms p99={:.3}ms max={:.3}ms", + v.len(), + p(0.5), + p(0.95), + p(0.99), + p(1.0) + ); +} + +async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) { + let mut v = Vec::new(); + let mut last = None; + for _ in 0..30 { + let t = Instant::now(); + let page = db.personal_read_sidebar(c, &actor.public_key(), 20, None).await.unwrap(); + v.push(t.elapsed()); + last = Some(page); + } + let page = last.unwrap(); + let busy = &page.channels.iter().max_by_key(|r| r.unread).unwrap(); + println!( + "BENCH {label}: busiest unread={} attention={} threads={} channels={}", + busy.unread, + busy.attention, + busy.threads.len(), + page.channels.len() + ); + stats(&format!("sidebar {label}"), v); +} + +#[tokio::test] +#[ignore = "benchmark"] +async fn bench_unread() { + let threads: usize = std::env::var("BENCH_THREADS").ok().and_then(|s| s.parse().ok()).unwrap_or(5000); + let pool = PgPoolOptions::new() + .max_connections(40) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool.clone()); + let c = db + .ensure_configured_community(&format!("bench-{}.local", Uuid::new_v4())) + .await + .unwrap() + .id; + let actor = Keys::generate(); + let mut channels = Vec::new(); + for i in 0..20 { + channels.push( + db.create_channel(c, &format!("bench {i}"), ChannelType::Stream, ChannelVisibility::Open, None, &actor.public_key().to_bytes(), None) + .await + .unwrap() + .id, + ); + } + let busy = channels[0]; + let others: Vec = (0..8).map(|_| Keys::generate()).collect(); + let base = ts() - 3600; + // A first message in every channel, and the first intent starts the actor. + let mut firsts = Vec::new(); + for ch in &channels { + let e = ev(&others[0], base, vec![]); + top(&db, c, *ch, &e).await; + firsts.push(e); + } + assert_eq!(apply(&db, c, &actor, channel_read(busy, &firsts[0])).await, IntentOutcome::Applied); + + // Phase 1: followed threads, serial ingest. + let (mut roots_t, mut reps_t) = (Vec::new(), Vec::new()); + let mut roots = Vec::new(); + let mut latest = firsts[0].clone(); + for i in 0..threads { + let at = base + 1 + (i as u64 % 3000); + let root = ev(&others[i % 8], at, vec![]); + roots_t.push(top(&db, c, busy, &root).await); + let mine = ev(&actor, at, vec![]); + reps_t.push(rep(&db, c, busy, &root, &mine).await); + for j in 1..3 { + let r = ev(&others[(i + j) % 8], at, vec![]); + reps_t.push(rep(&db, c, busy, &root, &r).await); + latest = r; + } + roots.push(root); + } + stats("ingest top-level serial", roots_t); + stats("ingest reply serial", reps_t); + assert_eq!(apply(&db, c, &actor, channel_read(busy, &latest)).await, IntentOutcome::Applied); + for (ch, e) in channels.iter().zip(&firsts).skip(1) { + apply(&db, c, &actor, channel_read(*ch, e)).await; + } + sqlx::query("ANALYZE").execute(&pool).await.unwrap(); + sidebar_bench(&db, c, &actor, &format!("caught-up {threads} threads")).await; + + // Phase 2: a 20k timeline backlog with 12 mentions. + let now = ts(); + let mut t2 = Vec::new(); + for i in 0..20_000u64 { + let tags = if i % 1667 == 0 { mention(&actor) } else { vec![] }; + let e = ev(&others[(i % 8) as usize], now - 600 + i / 40, tags); + t2.push(top(&db, c, busy, &e).await); + } + stats("ingest top-level serial (busy)", t2); + sqlx::query("ANALYZE").execute(&pool).await.unwrap(); + sidebar_bench(&db, c, &actor, "20k timeline backlog").await; + + // Phase 3: 500 followed threads get a new reply. + for root in roots.iter().take(500) { + let r = ev(&others[3], ts(), vec![]); + rep(&db, c, busy, root, &r).await; + } + sidebar_bench(&db, c, &actor, "20k backlog + 500 unread threads").await; + + // Concurrency: 16 writers into one channel, then into 16 channels. + for (label, spread) in [("same channel", false), ("16 channels", true)] { + let started = Instant::now(); + let mut handles = Vec::new(); + for w in 0..16usize { + let db = db.clone(); + let author = others[w % 8].clone(); + let ch = if spread { channels[w + 1] } else { busy }; + handles.push(tokio::spawn(async move { + let mut v = Vec::new(); + for i in 0..300u64 { + let e = ev(&author, ts() - (i % 30), vec![]); + v.push(top(&db, c, ch, &e).await); + } + v + })); + } + let mut all = Vec::new(); + for h in handles { + all.extend(h.await.unwrap()); + } + let n = all.len(); + let secs = started.elapsed().as_secs_f64(); + println!("BENCH concurrent {label}: {n} inserts in {secs:.2}s = {:.0}/s", n as f64 / secs); + stats(&format!("ingest concurrent {label}"), all); + } +} diff --git a/crates/buzz-db/src/store/personal_read/classification.rs b/crates/buzz-db/src/store/personal_read/classification.rs index 965519b8ad..f1238e3c70 100644 --- a/crates/buzz-db/src/store/personal_read/classification.rs +++ b/crates/buzz-db/src/store/personal_read/classification.rs @@ -3,7 +3,13 @@ use super::model::Reason; /// Why a message is directed, before conversation membership is known. -pub(super) fn reason(channel_type: &str, actor_hex: &str, tags: &[Vec]) -> Option { +/// `broadcast_reply` is a depth-1 reply broadcast to the channel timeline. +pub(super) fn reason( + channel_type: &str, + actor_hex: &str, + tags: &[Vec], + broadcast_reply: bool, +) -> Option { let tagged = |name: &str, matches: &dyn Fn(&str) -> bool| { tags.iter() .any(|tag| tag.len() >= 2 && tag[0] == name && matches(&tag[1])) @@ -12,7 +18,7 @@ pub(super) fn reason(channel_type: &str, actor_hex: &str, tags: &[Vec]) Some(Reason::Direct) } else if tagged("p", &|value| value.eq_ignore_ascii_case(actor_hex)) { Some(Reason::Mention) - } else if tagged("broadcast", &|value| value == "1") { + } else if broadcast_reply { Some(Reason::Broadcast) } else { None diff --git a/crates/buzz-db/src/store/personal_read/context.rs b/crates/buzz-db/src/store/personal_read/context.rs index 97fe7ebf0f..028cf78b6a 100644 --- a/crates/buzz-db/src/store/personal_read/context.rs +++ b/crates/buzz-db/src/store/personal_read/context.rs @@ -108,9 +108,9 @@ impl Db { let thread = row .try_get::>, _>("root_event_id")? .filter(|r| r != &message_id); - let on_timeline = thread.is_none() - || (row.try_get::, _>("depth")? == Some(1) - && row.try_get::, _>("broadcast")? == Some(true)); + let broadcast_reply = row.try_get::, _>("depth")? == Some(1) + && row.try_get::, _>("broadcast")? == Some(true); + let on_timeline = thread.is_none() || broadcast_reply; // A thread context holds its replies, not its root; a // broadcast reply belongs to both of its contexts. let in_context = if root.is_empty() { @@ -140,6 +140,7 @@ impl Db { &row.try_get::("channel_type")?, &actor_hex, &tags, + broadcast_reply, ); match position { None => MessageReadState::NotCounted, diff --git a/crates/buzz-db/src/store/personal_read/mod.rs b/crates/buzz-db/src/store/personal_read/mod.rs index 0c517f20ed..7dfd7e0b93 100644 --- a/crates/buzz-db/src/store/personal_read/mod.rs +++ b/crates/buzz-db/src/store/personal_read/mod.rs @@ -25,3 +25,6 @@ mod threads_postgres_tests; #[cfg(test)] mod arrival_postgres_tests; + +#[cfg(test)] +mod bench_postgres_tests; diff --git a/crates/buzz-db/src/store/personal_read/model.rs b/crates/buzz-db/src/store/personal_read/model.rs index 889fd5cc29..d013c8283b 100644 --- a/crates/buzz-db/src/store/personal_read/model.rs +++ b/crates/buzz-db/src/store/personal_read/model.rs @@ -51,8 +51,6 @@ pub enum IntentOutcome { pub const MAX_CHANNELS: usize = 20; /// Maximum unread-thread summaries per channel row. pub const MAX_THREAD_SUMMARIES: usize = 5; -/// Counts stop here: a count equal to the cap means at least the cap ("99+"). -pub const UNREAD_CAP: u32 = 99; /// Ingest rejects author times further than this from relay time, so anything /// that arrived after a position has an author time no earlier than this /// before it. Forward scans range over author time and filter on arrival. @@ -73,18 +71,15 @@ pub struct ChannelReadSummary { pub archived: bool, /// Existing DM visibility preference (not an authorization decision). pub hidden: bool, - /// Unread timeline messages plus unread replies in the actor's threads, - /// clamped at [`UNREAD_CAP`]. - pub unread: u32, - /// The unread subset with a [`Reason`]: everything but ordinary timeline - /// messages. Clamped at [`UNREAD_CAP`]. - pub attention: u32, - /// The actor's threads with at least one unread reply, clamped at - /// [`UNREAD_CAP`]. - pub unread_thread_count: u32, + /// An unread timeline message without a [`Reason`] exists: ordinary + /// backlog. Never true in a DM, where every message is direct. + pub unread: bool, + /// An unread message with a [`Reason`] exists: a direct, mention or + /// broadcast timeline message, or a reply in one of the actor's threads. + pub attention: bool, /// Last eligible nondeleted event to arrive among the channel's newest - /// and everything counted, whatever its author: `mark_channel_read` - /// through it reads the row. + /// and the unread messages found, whatever its author: + /// `mark_channel_read` through it reads the row. pub latest_message_id: Option, /// Display activity: the greatest author time (Unix seconds) among the /// channel's newest, not necessarily latest_message_id's own. None @@ -94,13 +89,11 @@ pub struct ChannelReadSummary { pub threads: Vec, } -/// Unread replies in one of the actor's threads. No conversation bytes. -#[derive(Debug, Serialize)] +/// One of the actor's threads with an unread reply. No conversation bytes. +#[derive(Debug, Deserialize, Serialize)] pub struct ThreadReadSummary { /// Canonical thread-root event ID. pub root_id: String, - /// Unread replies in this thread, clamped at [`UNREAD_CAP`]. - pub unread: u32, /// Last unread reply to arrive: marking through it reads the thread. pub latest_reply_id: String, /// Greatest author time (Unix seconds) among unread replies, not diff --git a/crates/buzz-db/src/store/personal_read/postgres_tests.rs b/crates/buzz-db/src/store/personal_read/postgres_tests.rs index 41b16a7376..5c09bb2998 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -236,24 +236,19 @@ async fn accounts(pool: &PgPool, community: CommunityId) -> i64 { #[tokio::test] #[ignore = "requires Postgres"] -async fn counts_stop_at_the_advertised_cap_and_serialize_as_integers() { +async fn unread_is_presence_and_serializes_as_booleans() { let (db, _, community, channel, actor, _) = fixture().await; let author = Keys::generate(); let base = now(); let mut last = None; - for i in 0..u64::from(UNREAD_CAP) + 20 { + for i in 0..120 { last = Some(post(&db, community, channel, &author, base + i, vec![]).await); } let row = sidebar(&db, community, &actor).await; - assert_eq!( - row.unread, UNREAD_CAP, - "a count at the cap means at least the cap" - ); let wire = serde_json::to_value(&row).unwrap(); - assert_eq!(wire["unread"], 99); - assert_eq!(wire["attention"], 0); - assert_eq!(wire["unread_thread_count"], 0); - for gone in ["latest_message_complete", "account", "status"] { + assert_eq!(wire["unread"], true); + assert_eq!(wire["attention"], false); + for gone in ["latest_message_complete", "account", "status", "unread_thread_count"] { assert!(wire.get(gone).is_none(), "{gone}"); } let last = last.unwrap(); @@ -262,14 +257,14 @@ async fn counts_stop_at_the_advertised_cap_and_serialize_as_integers() { apply(&db, community, &actor, mark(channel, None, &last)).await, IntentOutcome::Applied ); - assert_eq!(sidebar(&db, community, &actor).await.unread, 0); + assert!(!sidebar(&db, community, &actor).await.unread); } #[tokio::test] #[ignore = "requires Postgres"] async fn joining_starts_caught_up_and_counts_only_what_arrives_after() { let (db, pool, community, channel, owner, event) = fixture().await; - assert_eq!(sidebar(&db, community, &owner).await.unread, 1); + assert!(sidebar(&db, community, &owner).await.unread); let joiner = Keys::generate(); start(&pool, community, &joiner).await; let before = accounts(&pool, community).await; @@ -285,7 +280,7 @@ async fn joining_starts_caught_up_and_counts_only_what_arrives_after() { let row = sidebar(&db, community, &joiner).await; assert_eq!( (row.unread, row.attention), - (0, 0), + (false, false), "history before joining is read" ); assert_eq!(row.latest_message_id, Some(event.id.to_hex())); @@ -308,7 +303,7 @@ async fn joining_starts_caught_up_and_counts_only_what_arrives_after() { ) .await; let row = sidebar(&db, community, &joiner).await; - assert_eq!((row.unread, row.attention), (1, 1)); + assert_eq!((row.unread, row.attention), (false, true), "a mention is attention, not backlog"); assert_eq!( states(&db, community, &joiner, channel, None, &[&later]).await, [serde_json::json!({"status":"unread","reason":"mention"})] @@ -350,18 +345,18 @@ async fn rejoining_starts_caught_up_again() { ) .await .unwrap(); - assert_eq!(sidebar(&db, community, &owner).await.unread, 2); + assert!(sidebar(&db, community, &owner).await.unread); db.add_member(community, channel, &pk, MemberRole::Member, None) .await .unwrap(); let row = sidebar(&db, community, &member).await; - assert_eq!(row.unread, 0, "messages from while away are read"); + assert!(!row.unread, "messages from while away are read"); assert_eq!( status(&db, community, &member, channel, None, &[&away]).await, ["read"] ); post(&db, community, channel, &Keys::generate(), now(), vec![]).await; - assert_eq!(sidebar(&db, community, &member).await.unread, 1); + assert!(sidebar(&db, community, &member).await.unread); } #[tokio::test] @@ -408,8 +403,8 @@ async fn read_state_starts_caught_up_at_the_first_intent() { .await; let row = sidebar(&db, community, &member).await; assert_eq!( - (row.unread, row.attention, row.unread_thread_count), - (0, 0, 0) + (row.unread, row.attention, row.threads.len()), + (false, false, 0) ); assert_eq!(row.latest_message_id, Some(before.id.to_hex())); assert_eq!( @@ -423,7 +418,7 @@ async fn read_state_starts_caught_up_at_the_first_intent() { IntentOutcome::Applied ); let row = sidebar(&db, community, &member).await; - assert_eq!((row.unread, row.unread_thread_count), (0, 0)); + assert_eq!((row.unread, row.threads.len()), (false, 0)); assert_eq!( status(&db, community, &member, channel, None, &[&before]).await, ["read"] @@ -441,7 +436,7 @@ async fn read_state_starts_caught_up_at_the_first_intent() { .await; post(&db, community, channel, &Keys::generate(), now(), vec![]).await; let row = sidebar(&db, community, &member).await; - assert_eq!((row.unread, row.unread_thread_count), (2, 1)); + assert_eq!((row.unread, row.attention, row.threads.len()), (true, true, 1)); } #[tokio::test] @@ -461,7 +456,7 @@ async fn sidebar_is_read_only_and_does_not_wait_for_account() { .fetch_all(&mut *held) .await .unwrap(); - assert_eq!(sidebar(&db, community, &actor).await.unread, 0); + assert!(!sidebar(&db, community, &actor).await.unread); held.rollback().await.unwrap(); } @@ -539,7 +534,7 @@ async fn diff_alone_leaves_the_sidebar_row_unchanged() { ) .await; let after = serde_json::to_value(sidebar(&db, community, &actor).await).unwrap(); - assert_eq!(before["unread"], 1); + assert_eq!(before["unread"], true); assert_eq!(before, after, "not unread, not attention, not latest"); } @@ -560,7 +555,7 @@ async fn latest_includes_own_and_excludes_deleted_and_auxiliary() { let row = sidebar(&db, community, &actor).await; assert_eq!(row.latest_message_id, Some(own.id.to_hex())); assert_eq!(row.latest_message_at, Some((base + 1) as i64)); - assert_eq!(row.unread, 1, "own, deleted and reaction do not count"); + assert!(row.unread, "own, deleted and reaction do not count"); // An empty channel has no latest message. db.create_channel( community, @@ -621,16 +616,15 @@ async fn position_is_monotonic_and_rejects_malformed_anchors() { vec![], ) .await; - assert_eq!(sidebar(&db, community, &actor).await.unread, 1); + assert!(sidebar(&db, community, &actor).await.unread); for anchor in [&later, &event] { assert_eq!( apply(&db, community, &actor, mark(channel, None, anchor)).await, IntentOutcome::Applied ); } - assert_eq!( - sidebar(&db, community, &actor).await.unread, - 0, + assert!( + !sidebar(&db, community, &actor).await.unread, "never moves back" ); let at_later: bool = sqlx::query_scalar( @@ -684,7 +678,7 @@ async fn contexts_bound_selectors_and_share_one_state_per_message() { status(&db, community, &actor, channel, Some(&root), &both).await, ["unavailable", "unread", "not_counted"] ); - assert_eq!(sidebar(&db, community, &actor).await.unread, 2); + assert!(sidebar(&db, community, &actor).await.unread); assert_eq!( accounts(&pool, community).await, 1, diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index ea47e9bd67..78cfc0928d 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -1,10 +1,9 @@ -//! Forward counting from read positions. A caught-up reader examines almost -//! nothing: each count reads only what arrived after its position, and stops -//! at [`UNREAD_CAP`]. +//! Unread presence from read positions. Every fact is an existence probe +//! that stops at the first qualifying row, so a caught-up reader examines +//! almost nothing and nothing is counted. use super::{model::*, writes}; use buzz_core::CommunityId; -use serde::Deserialize; use sqlx::{Acquire, Row}; use uuid::Uuid; @@ -16,9 +15,13 @@ use crate::{observability, Db, DbError, Result}; const ON_TIMELINE: &str = "(tm.root_event_id IS NULL OR tm.root_event_id=e.id OR (tm.depth=1 AND tm.broadcast))"; -/// Whether event `e` is directed at the actor (hex `$hex`) by its tags. -const DIRECTED_TAGS: &str = "EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag) - WHERE (tag->>0='p' AND lower(tag->>1)=$hex) OR (tag->>0='broadcast' AND tag->>1='1'))"; +/// Event `e` is eligible and unread against `r.position`, and `tm` is its +/// thread row. Author time ranges from the position less the arrival skew. +const UNREAD_EVENT: &str = "e.created_at >= r.position-$8 AND e.received_at > r.position + AND e.kind=ANY($6) AND e.pubkey<>$2 AND e.deleted_at IS NULL"; + +const EVENT_THREAD_ROW: &str = "LEFT JOIN thread_metadata tm ON tm.community_id=$1 + AND tm.event_created_at=e.created_at AND tm.event_id=e.id"; impl Db { /// Read a bounded joined roster from the writer in one read-only snapshot. @@ -75,13 +78,18 @@ impl Db { // the actor's first intent, so nothing counts). A channel's is its // frontier, never earlier than joining: a new member starts caught // up. A thread row exists only for the actor's threads (see - // `membership`), and a whole-channel cut covers them all. Each count - // ranges over author time from the position less the arrival skew - // (served by idx_events_community_channel_created or - // idx_thread_metadata_window), filters on arrival, and stops at the - // cap: rows the filters reject are read but bound nothing. The latest message is the last arrival among - // the newest messages and everything counted, so marking through it - // reads every counted message whatever its author time. + // `membership`), and a whole-channel cut covers them all. + // + // Each fact is the first qualifying row of one index range: + // - ordinary: the channel's timeline after the position, skipping + // directed messages (idx_events_community_channel_created); + // - mention: the actor's own mentions (idx_event_mentions_pubkey_created); + // - broadcast: the channel's depth-1 replies (idx_thread_metadata_channel_depth); + // - direct: any unread timeline message in a DM; + // - threads: roots in the channel whose last reply arrived after the + // actor's position, then their unread replies (anchor and display time). + // The latest message is the last arrival among the newest messages and + // every unread message found, so marking through it reads what was found. let sql = format!( r#"WITH roster AS MATERIALIZED ( SELECT c.id, c.name, c.channel_type::text AS channel_type, @@ -100,11 +108,12 @@ impl Db { ORDER BY c.id LIMIT $5 ) SELECT r.id, r.name, r.channel_type, r.archived, r.hidden, - (SELECT v.id FROM (VALUES (latest.arrival, latest.id), - (timeline.arrival, timeline.id), (threads.arrival, threads.id)) v(arrival, id) + (SELECT v.id FROM (VALUES (latest.arrival, latest.id), (ordinary.arrival, ordinary.id), + (directed.arrival, directed.id), (threads.arrival, threads.id)) v(arrival, id) WHERE v.id IS NOT NULL ORDER BY v.arrival DESC, v.id LIMIT 1) AS latest_message_id, latest.at AS latest_message_at, - COALESCE(timeline.unread, 0) AS unread, COALESCE(timeline.attention, 0) AS attention, + ordinary.id IS NOT NULL AS unread, + directed.id IS NOT NULL OR threads.id IS NOT NULL AS attention, COALESCE(threads.items, '[]') AS threads FROM roster r LEFT JOIN LATERAL ( @@ -115,49 +124,68 @@ impl Db { ORDER BY created_at DESC LIMIT $7) newest ) latest ON true LEFT JOIN LATERAL ( - SELECT count(*)::int AS unread, count(*) FILTER (WHERE directed)::int AS attention, - (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id, - max(received_at) AS arrival - FROM ( - SELECT e.id, e.received_at, r.channel_type='dm' OR {DIRECTED_TAGS} AS directed - FROM events e - LEFT JOIN thread_metadata tm ON tm.community_id=$1 - AND tm.event_created_at=e.created_at AND tm.event_id=e.id - WHERE e.community_id=$1 AND e.channel_id=r.id - AND e.created_at >= r.position-$8 AND e.received_at > r.position - AND e.kind=ANY($6) AND e.pubkey<>$2 AND e.deleted_at IS NULL AND {ON_TIMELINE} - ORDER BY e.created_at LIMIT $9 - ) counted - ) timeline ON true + SELECT encode(e.id,'hex') AS id, e.received_at AS arrival + FROM events e {EVENT_THREAD_ROW} + WHERE r.channel_type<>'dm' AND e.community_id=$1 AND e.channel_id=r.id + AND {UNREAD_EVENT} AND {ON_TIMELINE} + AND NOT (tm.depth=1 AND tm.broadcast) IS TRUE + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag) + WHERE tag->>0='p' AND lower(tag->>1)=$hex) + ORDER BY e.created_at LIMIT 1 + ) ordinary ON true + LEFT JOIN LATERAL ( + SELECT id, arrival FROM ( + (SELECT encode(e.id,'hex') AS id, e.received_at AS arrival + FROM events e {EVENT_THREAD_ROW} + WHERE r.channel_type='dm' AND e.community_id=$1 AND e.channel_id=r.id + AND {UNREAD_EVENT} AND {ON_TIMELINE} + ORDER BY e.created_at LIMIT 1) + UNION ALL + (SELECT encode(e.id,'hex'), e.received_at + FROM event_mentions m JOIN events e ON e.community_id=$1 + AND e.created_at=m.event_created_at AND e.id=m.event_id + {EVENT_THREAD_ROW} + WHERE r.channel_type<>'dm' AND m.community_id=$1 AND m.pubkey_hex=$hex + AND m.event_created_at >= r.position-$8 AND m.channel_id=r.id + AND e.channel_id=r.id AND {UNREAD_EVENT} AND {ON_TIMELINE} + LIMIT 1) + UNION ALL + (SELECT encode(e.id,'hex'), e.received_at + FROM thread_metadata tm JOIN events e ON e.community_id=$1 + AND e.created_at=tm.event_created_at AND e.id=tm.event_id + WHERE r.channel_type<>'dm' AND tm.community_id=$1 AND tm.channel_id=r.id + AND tm.depth=1 AND tm.broadcast AND tm.event_created_at >= r.position-$8 + AND {UNREAD_EVENT} + LIMIT 1) + ) found ORDER BY arrival DESC LIMIT 1 + ) directed ON true LEFT JOIN LATERAL ( SELECT jsonb_agg(jsonb_build_object('root_id',encode(tf.root_id,'hex'), - 'unread',n.unread,'latest_reply_id',n.id,'latest_reply_at',n.at)) AS items, + 'latest_reply_id',n.id,'latest_reply_at',n.at)) AS items, (array_agg(n.id ORDER BY n.arrival DESC,n.id))[1] AS id, max(n.arrival) AS arrival FROM personal_read_frontiers tf + JOIN thread_metadata root ON root.community_id=$1 AND root.channel_id=r.id + AND root.depth=0 AND root.event_id=tf.root_id CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, r.threads_floor) AS position) p CROSS JOIN LATERAL ( - SELECT count(*)::int AS unread, - (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id, - max(extract(epoch FROM created_at)::bigint) AS at, - max(received_at) AS arrival - FROM ( - SELECT e.id, e.created_at, e.received_at - FROM thread_metadata tm JOIN events e ON e.community_id=$1 - AND e.created_at=tm.event_created_at AND e.id=tm.event_id - WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id - AND tm.channel_id=r.id AND tm.event_id<>tf.root_id - AND tm.event_created_at >= p.position-$8 - AND e.received_at > p.position AND e.kind=ANY($6) AND e.pubkey<>$2 - AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast) - ORDER BY tm.event_created_at LIMIT $9 - ) counted + SELECT (array_agg(encode(e.id,'hex') ORDER BY e.received_at DESC,e.id))[1] AS id, + max(extract(epoch FROM e.created_at)::bigint) AS at, + max(e.received_at) AS arrival + FROM thread_metadata tm JOIN events e ON e.community_id=$1 + AND e.created_at=tm.event_created_at AND e.id=tm.event_id + WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id + AND tm.channel_id=r.id AND tm.event_id<>tf.root_id + AND tm.event_created_at >= p.position-$8 + AND e.received_at > p.position AND e.kind=ANY($6) AND e.pubkey<>$2 + AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast) ) n WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id - AND tf.root_id<>''::bytea AND n.unread > 0 + AND tf.root_id<>''::bytea AND root.last_reply_received_at > p.position + AND n.id IS NOT NULL ) threads ON true ORDER BY r.id"# ) - .replace("$hex", "$10"); + .replace("$hex", "$9"); // Only compile-time constants are interpolated. let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(community.as_uuid()) @@ -168,7 +196,6 @@ impl Db { .bind(ELIGIBLE_KINDS.as_slice()) .bind(LATEST_PROBE as i64) .bind(skew()) - .bind(i64::from(UNREAD_CAP)) .bind(actor.to_hex()) .fetch_all(&mut *tx) .await?; @@ -176,23 +203,19 @@ impl Db { let has_more = rows.len() > limit; let mut channels = Vec::with_capacity(limit); for row in rows.into_iter().take(limit) { - let threads: Vec = serde_json::from_value(row.try_get("threads")?) + let threads: Vec = serde_json::from_value(row.try_get("threads")?) .map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?; - let thread_unread: u32 = threads.iter().map(|t| t.unread).sum(); - let unread = row.try_get::("unread")? as u32; - let attention = row.try_get::("attention")? as u32; channels.push(ChannelReadSummary { channel_id: row.try_get("id")?, name: row.try_get("name")?, channel_type: row.try_get("channel_type")?, archived: row.try_get("archived")?, hidden: row.try_get("hidden")?, - unread: clamp(unread + thread_unread), - attention: clamp(attention + thread_unread), - unread_thread_count: clamp(threads.len() as u32), + unread: row.try_get("unread")?, + attention: row.try_get("attention")?, latest_message_id: row.try_get("latest_message_id")?, latest_message_at: row.try_get("latest_message_at")?, - threads: summarize(threads.into_iter().map(ThreadRow::into).collect()), + threads: summarize(threads), }); } let next_cursor = if has_more { @@ -214,29 +237,6 @@ pub(super) fn skew() -> chrono::Duration { chrono::Duration::seconds(i64::from(MAX_ARRIVAL_SKEW_SECONDS)) } -fn clamp(n: u32) -> u32 { - n.min(UNREAD_CAP) -} - -#[derive(Deserialize)] -struct ThreadRow { - root_id: String, - unread: u32, - latest_reply_id: String, - latest_reply_at: i64, -} - -impl From for ThreadReadSummary { - fn from(t: ThreadRow) -> Self { - Self { - root_id: t.root_id, - unread: clamp(t.unread), - latest_reply_id: t.latest_reply_id, - latest_reply_at: t.latest_reply_at, - } - } -} - /// Order newest unread reply first (root ID breaks ties) and cap the list. pub(super) fn summarize(mut items: Vec) -> Vec { items.sort_unstable_by(|a, b| { @@ -255,7 +255,6 @@ mod tests { fn item(root: u8, at: i64) -> ThreadReadSummary { ThreadReadSummary { root_id: hex::encode([root; 32]), - unread: 1, latest_reply_id: hex::encode([root; 32]), latest_reply_at: at, } diff --git a/crates/buzz-db/src/store/personal_read/projection_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/projection_postgres_tests.rs index 8a1436cb1f..918502424d 100644 --- a/crates/buzz-db/src/store/personal_read/projection_postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/projection_postgres_tests.rs @@ -26,7 +26,7 @@ async fn sidebar_sql_eligibility_matches_selector() { .unwrap(); let expected = u32::from(ELIGIBLE_KINDS.contains(&kind) && !own && !deleted); assert_eq!( - page.channels[0].unread, expected, + u32::from(page.channels[0].unread), expected, "kind={kind} own={own} deleted={deleted}" ); let contexts = db @@ -47,7 +47,7 @@ async fn sidebar_sql_eligibility_matches_selector() { #[tokio::test] #[ignore = "requires Postgres"] async fn sidebar_directed_fact_matches_selector_classifier() { - let (db, pool, community, channel, actor, _) = fixture().await; + let (db, pool, community, channel, actor, event) = fixture().await; let actor_hex = actor.public_key().to_hex(); let fullwidth: String = actor_hex .chars() @@ -79,14 +79,36 @@ async fn sidebar_directed_fact_matches_selector_classifier() { .await .unwrap(); for tags in &cases { - let expected = - u32::from(classification::reason(channel_type, &actor_hex, tags).is_some()); + // A top-level message: `broadcast` tags alone direct nothing. + let reason = classification::reason(channel_type, &actor_hex, tags, false); + let expected = (channel_type != "dm" && reason.is_none(), reason.is_some()); sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1") .bind(community.as_uuid()) .bind(json!(tags)) .execute(&pool) .await .unwrap(); + // Mirror ingest's mention index (`runtime::insert_mentions`). + sqlx::query("DELETE FROM event_mentions WHERE community_id=$1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .unwrap(); + let mentioned = tags.iter().any(|t| { + t.len() >= 2 && t[0] == "p" && t[1].eq_ignore_ascii_case(&actor_hex) + }); + if mentioned { + sqlx::query( + "INSERT INTO event_mentions (community_id,pubkey_hex,event_id,event_created_at,channel_id,event_kind) + SELECT community_id,$2,id,created_at,channel_id,kind FROM events WHERE community_id=$1 AND id=$3", + ) + .bind(community.as_uuid()) + .bind(&actor_hex) + .bind(event.id.as_bytes().as_slice()) + .execute(&pool) + .await + .unwrap(); + } let row = db .personal_read_sidebar(community, &actor.public_key(), 20, None) .await @@ -95,7 +117,7 @@ async fn sidebar_directed_fact_matches_selector_classifier() { .remove(0); assert_eq!( (row.unread, row.attention), - (1, expected), + expected, "channel_type={channel_type} tags={tags:?}" ); } diff --git a/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs index 3552d2273c..c6fe011cbe 100644 --- a/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs @@ -65,9 +65,10 @@ async fn replying_mentioning_and_starting_a_thread_make_members() { assert_eq!(thread_rows(&pool, community, &starter).await, root_row); assert!(thread_rows(&pool, community, &bystander).await.is_empty()); let row = sidebar(&db, community, &starter).await; + // Only the reply is unread: attention, but no ordinary timeline backlog. assert_eq!( - (row.unread, row.attention, row.unread_thread_count), - (1, 1, 1) + (row.unread, row.attention, row.threads.len()), + (false, true, 1) ); assert_eq!(row.threads[0].latest_reply_id, own.id.to_hex()); assert_eq!( @@ -75,7 +76,7 @@ async fn replying_mentioning_and_starting_a_thread_make_members() { [serde_json::json!({"status":"unread","reason":"conversation"})] ); let row = sidebar(&db, community, &actor).await; - assert_eq!((row.unread_thread_count, row.threads.len()), (0, 0)); + assert!(row.threads.is_empty()); // Someone else replies: the actor, now a member, has it unread, beside // the fixture's message and the root on the timeline. @@ -92,10 +93,9 @@ async fn replying_mentioning_and_starting_a_thread_make_members() { .await; let row = sidebar(&db, community, &actor).await; assert_eq!( - (row.unread, row.attention, row.unread_thread_count), - (3, 1, 1) + (row.unread, row.attention, row.threads.len()), + (true, true, 1) ); - assert_eq!(row.threads[0].unread, 1); assert_eq!(row.threads[0].latest_reply_id, other.id.to_hex()); // A mention makes a channel member a thread member; a non-member gets nothing. @@ -118,8 +118,8 @@ async fn replying_mentioning_and_starting_a_thread_make_members() { let row = sidebar(&db, community, &bystander).await; // The mention, and the root on the timeline. assert_eq!( - (row.unread, row.attention, row.unread_thread_count), - (2, 1, 1) + (row.unread, row.attention, row.threads.len()), + (true, true, 1) ); assert_eq!( states( @@ -200,13 +200,13 @@ async fn reading_a_thread_moves_a_members_row_and_never_joins() { false, ) .await; - assert_eq!(sidebar(&db, community, &actor).await.threads[0].unread, 1); + assert_eq!(sidebar(&db, community, &actor).await.threads.len(), 1); assert_eq!( apply(&db, community, &actor, mark(channel, Some(&root), &root)).await, IntentOutcome::Applied ); assert_eq!( - sidebar(&db, community, &actor).await.unread_thread_count, + sidebar(&db, community, &actor).await.threads.len(), 1, "never backwards" ); @@ -216,8 +216,8 @@ async fn reading_a_thread_moves_a_members_row_and_never_joins() { ); let row = sidebar(&db, community, &actor).await; assert_eq!( - (row.unread, row.unread_thread_count), - (1, 0), + (row.unread, row.threads.len()), + (true, 0), "the root itself is on the timeline" ); } @@ -263,8 +263,8 @@ async fn dm_replies_make_every_member_a_thread_member() { .channels .remove(0); assert_eq!( - (row.unread, row.attention, row.unread_thread_count), - (2, 2, 1) + (row.unread, row.attention, row.threads.len()), + (false, true, 1) ); } @@ -316,7 +316,11 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() { IntentOutcome::Applied ); let row = sidebar(&db, community, &actor).await; - assert_eq!(row.unread, 2, "top at +20 and reply at +30 remain"); + assert_eq!( + (row.unread, row.attention), + (true, true), + "top at +20 and reply at +30 remain" + ); assert_eq!(row.threads.len(), 1); assert_eq!(row.threads[0].latest_reply_id, second.id.to_hex()); @@ -326,8 +330,8 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() { ); let row = sidebar(&db, community, &actor).await; assert_eq!( - (row.unread, row.unread_thread_count), - (0, 0), + (row.unread, row.attention, row.threads.len()), + (false, false, 0), "every thread is covered" ); @@ -355,7 +359,7 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() { ) .await; let row = sidebar(&db, community, &actor).await; - assert_eq!(row.unread, 2); + assert_eq!((row.unread, row.attention), (false, true)); assert_eq!(row.threads[0].latest_reply_id, newer.id.to_hex()); assert_eq!( status( @@ -453,11 +457,7 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() { } let row = sidebar(&db, community, &actor).await; // The fixture's message is one ordinary timeline unread. - assert_eq!((row.unread, row.attention), (9, 8)); - assert_eq!( - row.unread_thread_count, 6, - "counts threads beyond the listed five" - ); + assert_eq!((row.unread, row.attention, row.threads.len()), (true, true, 5)); let mut tied = [roots[0].id.to_hex(), roots[1].id.to_hex()]; tied.sort(); let order: Vec<_> = row.threads.iter().map(|t| t.root_id.clone()).collect(); @@ -471,7 +471,6 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() { roots[4].id.to_hex() ] ); - assert_eq!(row.threads[2].unread, 3); assert_eq!(row.threads[2].latest_reply_id, last.unwrap().id.to_hex()); assert_eq!(row.threads[2].latest_reply_at, (base + 140) as i64); // The row's activity is the greatest author time among the newest messages. @@ -491,12 +490,10 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() { IntentOutcome::Applied ); let row = sidebar(&db, community, &actor).await; - assert_eq!((row.unread_thread_count, row.threads.len()), (5, 5)); + assert_eq!(row.threads.len(), 5); + assert!(row.threads.iter().all(|t| t.root_id != first.root_id)); // Plus the fixture's timeline message. - assert_eq!( - row.unread, - 1 + row.threads.iter().map(|t| t.unread).sum::() - ); + assert!(row.unread); } #[tokio::test] @@ -539,7 +536,7 @@ async fn thread_on_a_never_unread_root_is_selectable_and_readable_by_itself() { .await; let row = sidebar(&db, community, &actor).await; - assert_eq!(row.unread, 3, "the root and two replies"); + assert_eq!((row.unread, row.attention), (true, true), "the root and two replies"); assert_eq!(row.threads.len(), 2); let listed = &row.threads[1]; assert_eq!(listed.root_id, diff.id.to_hex()); @@ -562,7 +559,7 @@ async fn thread_on_a_never_unread_root_is_selectable_and_readable_by_itself() { IntentOutcome::Applied ); let row = sidebar(&db, community, &actor).await; - assert_eq!(row.unread, 2); + assert_eq!((row.unread, row.attention), (true, true)); assert_eq!(row.threads.len(), 1); assert_eq!(row.threads[0].latest_reply_id, elsewhere.id.to_hex()); } diff --git a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs index e19614d0d3..07121c4569 100644 --- a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs +++ b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs @@ -464,7 +464,7 @@ async fn accessory_discovery_is_host_bound_and_opt_in() { let d = &doc["buzz_v1"]; assert_eq!(d["version"], 1); assert_eq!(d["base_path"], "/buzz/v1"); - assert_eq!(d["unread_cap"], 99); + assert!(d.get("unread_cap").is_none()); assert!(d.get("retention_seconds").is_none()); assert_eq!(d["max_channels"], buzz_db::personal_read::MAX_CHANNELS); assert_eq!(d["max_intents"], buzz_db::personal_read::MAX_INTENTS); diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index a13d70c509..d0aa81f54a 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -88,8 +88,6 @@ pub struct BuzzV1Descriptor { pub version: u32, /// Relay-relative API prefix; callers retain the requesting origin. pub base_path: String, - /// Unread counts stop here: a count equal to the cap means at least it. - pub unread_cap: u32, /// Maximum joined channels per sidebar page. pub max_channels: usize, /// Maximum independent write intents per request. @@ -391,12 +389,11 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st if state.config.buzz_v1_enabled { use buzz_db::personal_read::{ ELIGIBLE_KINDS, MAX_CHANNELS, MAX_CONTEXTS, MAX_CONTEXT_MESSAGES, MAX_INTENTS, - MAX_THREAD_SUMMARIES, UNREAD_CAP, + MAX_THREAD_SUMMARIES, }; info.buzz_v1 = Some(BuzzV1Descriptor { version: 1, base_path: crate::api::buzz_v1::BASE_PATH.to_owned(), - unread_cap: UNREAD_CAP, max_channels: MAX_CHANNELS, max_intents: MAX_INTENTS, max_contexts: MAX_CONTEXTS, diff --git a/migrations/0056_personal_read_state.sql b/migrations/0056_personal_read_state.sql index aa06f16489..3b5ef3a7a6 100644 --- a/migrations/0056_personal_read_state.sql +++ b/migrations/0056_personal_read_state.sql @@ -32,6 +32,9 @@ CREATE TABLE personal_read_frontiers ( ); +-- Arrival time of a thread root's latest reply: the sidebar skips threads +-- with nothing new since the reader's position without reading replies. +ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ; SELECT attach_community_write_fence('personal_read_accounts'); SELECT attach_community_write_fence('personal_read_frontiers'); diff --git a/schema/schema.sql b/schema/schema.sql index a54f877671..1f3de52f57 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -559,6 +559,7 @@ CREATE TABLE thread_metadata ( descendant_count INT NOT NULL DEFAULT 0, last_reply_at TIMESTAMPTZ, broadcast BOOLEAN NOT NULL DEFAULT FALSE, + last_reply_received_at TIMESTAMPTZ, PRIMARY KEY (community_id, event_created_at, event_id), FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id) );