diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 4d28503edd..e77d651592 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -107,9 +107,17 @@ pub(crate) async fn insert_mentions_in_transaction( // transaction so all chunks share its commit boundary. const MENTION_INSERT_CHUNK_ROWS: usize = 5_000; for chunk in valid_pubkeys.chunks(MENTION_INSERT_CHUNK_ROWS) { + // Arrival time and read scope come from the stored event and its + // thread row, which every caller writes before (or with) its mentions. let mut qb: QueryBuilder = QueryBuilder::new( "INSERT INTO event_mentions \ - (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) ", + (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind, \ + received_at, root_id) \ + SELECT v.community_id, v.pubkey_hex, v.event_id, v.event_created_at, v.channel_id, \ + v.event_kind, e.received_at, \ + CASE WHEN tm.root_event_id <> v.event_id AND NOT (tm.depth = 1 AND tm.broadcast) \ + THEN tm.root_event_id END \ + FROM (", ); qb.push_values(chunk, |mut b, pubkey| { @@ -121,7 +129,14 @@ pub(crate) async fn insert_mentions_in_transaction( .push_bind(kind as i32); }); - qb.push(" ON CONFLICT DO NOTHING"); + qb.push( + ") v (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) \ + LEFT JOIN events e ON e.community_id = v.community_id \ + AND e.created_at = v.event_created_at AND e.id = v.event_id \ + LEFT JOIN thread_metadata tm ON tm.community_id = v.community_id \ + AND tm.event_created_at = v.event_created_at AND tm.event_id = v.event_id \ + ON CONFLICT DO NOTHING", + ); qb.build().execute(&mut **tx).await?; } diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index fa6feccb2a..eb3294761c 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -444,6 +444,17 @@ pub async fn insert_event_in_transaction( ) -> Result<(StoredEvent, bool)> { let result = insert_event_on(tx.as_mut(), community_id, event, channel_id).await?; if result.1 { + if let Some(channel) = channel_id { + crate::personal_read::record_message( + tx, + community_id, + event, + channel, + crate::personal_read::Place::TopLevel, + result.0.received_at, + ) + .await?; + } crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?; } Ok(result) @@ -1853,12 +1864,15 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( .bind(received_at) .execute(&mut **tx) .await?; - crate::personal_read::record_reply( + crate::personal_read::record_message( tx, community_id, event, meta.channel_id, - root, + crate::personal_read::Place::Reply { + root, + broadcast: meta.depth == 1 && meta.broadcast, + }, received_at, ) .await?; @@ -1866,6 +1880,17 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( } } + if let (None, Some(channel)) = (&thread_meta, channel_id) { + crate::personal_read::record_message( + tx, + community_id, + event, + channel, + crate::personal_read::Place::TopLevel, + received_at, + ) + .await?; + } crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?; } 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 ef3fd98905..3fb4034113 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 @@ -1,6 +1,6 @@ //! Read progress follows relay arrival (`received_at`), never author time. //! Each case sets every arrival explicitly: back-to-back inserts share a clock. -use super::{postgres_tests, projection::LATEST_PROBE, *}; +use super::{postgres_tests, *}; use crate::Db; use buzz_core::CommunityId; use nostr::{EventBuilder, Keys, Kind, Tag}; @@ -182,30 +182,82 @@ async fn future_dated_anchor_does_not_swallow_later_arrivals() { assert!(sidebar(&db, community, &actor).await.unread); } -/// Marking the sidebar's own latest message must clear the badge even when the -/// last arrival is not the newest by author time. +/// Clients name only display-order IDs. Marking the newest message shown +/// must clear the badge even when the last arrival is displayed above it. #[tokio::test] #[ignore = "requires Postgres"] -async fn mark_as_read_with_the_sidebar_anchor_clears_a_late_arrival() { +async fn marking_the_newest_displayed_message_clears_a_late_arrival_above_it() { let (db, pool, community, channel, actor, first) = fixture().await; let now = first.created_at.as_secs(); 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; + let newest = post(&db, &pool, community, channel, now + 1, now - 40).await; + let late = post(&db, &pool, community, channel, now - 600, now - 30).await; assert!(sidebar(&db, community, &actor).await.unread); - let anchor = sidebar(&db, community, &actor) - .await - .latest_message_id - .expect("a channel with messages has a latest message"); - apply(&db, community, &actor, mark_channel_read(channel, anchor)).await; + apply( + &db, + community, + &actor, + mark_through(channel, None, &newest.id.to_hex()), + ) + .await; + let row = sidebar(&db, community, &actor).await; + assert!(!row.unread); + assert_eq!( + row.read_through_id, + Some(newest.id.to_hex()), + "the divider stays under the newest displayed message" + ); + assert_eq!( + states(&db, community, &actor, channel, &[&late, &newest]).await, + ["read", "read"] + ); +} + +/// A mark reads only what is displayed at or before its anchor: a message +/// shown below it stays unread, even one that arrived first. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn marking_leaves_messages_displayed_below_the_anchor_unread() { + let (db, pool, community, channel, actor, first) = fixture().await; + let now = first.created_at.as_secs(); + arrive(&pool, community, &first, now - 60).await; + let below = post(&db, &pool, community, channel, now + 5, now - 50).await; + apply( + &db, + community, + &actor, + mark_through(channel, None, &first.id.to_hex()), + ) + .await; + assert_eq!( + states(&db, community, &actor, channel, &[&first, &below]).await, + ["read", "unread"] + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn mark_channel_read_with_the_newest_displayed_message_clears_a_late_arrival() { + let (db, pool, community, channel, actor, first) = fixture().await; + let now = first.created_at.as_secs(); + arrive(&pool, community, &first, now - 60).await; + let newest = post(&db, &pool, community, channel, now + 1, now - 40).await; + post(&db, &pool, community, channel, now - 600, now - 30).await; + apply( + &db, + community, + &actor, + mark_channel_read(channel, newest.id.to_hex()), + ) + .await; assert!(!sidebar(&db, community, &actor).await.unread); } #[tokio::test] #[ignore = "requires Postgres"] -async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { +async fn mark_thread_read_with_the_newest_displayed_reply_clears_a_late_reply() { let (db, pool, community, channel, actor, root) = fixture().await; let now = root.created_at.as_secs(); arrive(&pool, community, &root, now - 60).await; @@ -216,7 +268,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { mark_through(channel, None, &root.id.to_hex()), ) .await; - reply( + let newest = reply( &db, &pool, community, @@ -227,7 +279,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { now - 40, ) .await; - reply( + let late = reply( &db, &pool, community, @@ -248,80 +300,25 @@ 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.attention, row.threads.len()), (false, true, 1)); - assert_eq!(row.threads.len(), 1); + assert_eq!((row.unread, row.threads.len()), (false, 1)); + assert_eq!( + row.threads[0].latest_id, + late.id.to_hex(), + "the last arrival" + ); + assert_eq!(row.threads[0].mentions, 2); let root_id = root.id.to_hex(); - let anchor = row.threads[0].latest_reply_id.clone(); apply( &db, community, &actor, - mark_through(channel, Some(&root_id), &anchor), + mark_through(channel, Some(&root_id), &newest.id.to_hex()), ) .await; let row = sidebar(&db, community, &actor).await; - assert_eq!((row.unread, row.attention, row.threads.len()), (false, false, 0)); - assert!(row.threads.is_empty()); -} - -/// The latest probe holds only the newest messages by author time. A late -/// arrival authored before all of them is still counted, so it must still be -/// the anchor, or Mark as read with the sidebar anchor leaves it unread. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn mark_as_read_clears_a_late_arrival_behind_the_latest_probe() { - let (db, pool, community, channel, actor, first) = fixture().await; - let now = first.created_at.as_secs(); - arrive(&pool, community, &first, now - 60).await; - for i in 0..LATEST_PROBE as u64 { - post(&db, &pool, community, channel, now - 500 + i, now - 50).await; - } - let anchor = sidebar(&db, community, &actor) - .await - .latest_message_id - .unwrap(); - apply(&db, community, &actor, mark_channel_read(channel, anchor)).await; - 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!(row.unread); - assert_eq!(row.latest_message_id, Some(late.id.to_hex())); - apply( - &db, - community, - &actor, - mark_channel_read(channel, late.id.to_hex()), - ) - .await; - assert!(!sidebar(&db, community, &actor).await.unread); -} - -/// Reactions never fill the latest probe. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn latest_message_behind_newer_reactions_is_found() { - let (db, pool, community, channel, actor, first) = fixture().await; - let now = first.created_at.as_secs(); - let demoted = sqlx::query("UPDATE events SET kind=7 WHERE community_id=$1 AND id=$2") - .bind(community.as_uuid()) - .bind(first.id.as_bytes().as_slice()) - .execute(&pool) - .await - .unwrap() - .rows_affected(); - assert_eq!(demoted, 1); - let only = post(&db, &pool, community, channel, now - 600, now - 10).await; - for i in 0..LATEST_PROBE as u64 { - post_kind(&db, &pool, community, channel, 7, now - 500 + i, now - 50).await; - } - - 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!(row.unread); + assert_eq!((row.unread, row.threads.len()), (false, 0)); } /// Store `event` as having arrived at exactly `seconds` plus `micros`. Built @@ -359,10 +356,13 @@ async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec, 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; + // Equal author times display in ID order; index 0 displays first, so a + // mark reads only what arrived by then, not the other by display order. + let mut both = [&first, &second]; + both.sort_by_key(|e| e.id); let arrived = now as i64 - 30; - arrive_exact(&pool, community, &first, arrived, micros[0]).await; - arrive_exact(&pool, community, &second, arrived, micros[1]).await; - let both = [&first, &second]; + arrive_exact(&pool, community, both[0], arrived, micros[0]).await; + arrive_exact(&pool, community, both[1], arrived, micros[1]).await; apply( &db, community, 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 index dd34fb7697..a10689d373 100644 --- a/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs @@ -2,11 +2,11 @@ //! 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 buzz_core::CommunityId; use nostr::{EventBuilder, Keys, Kind, Tag}; use sqlx::postgres::PgPoolOptions; use std::time::{Duration, Instant}; @@ -31,7 +31,9 @@ async fn top(db: &Db, c: CommunityId, ch: Uuid, e: &nostr::Event) -> Duration { } 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 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, @@ -72,16 +74,19 @@ async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) { 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(); + 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={}", + "BENCH {label}: busiest unread={} mentions={} threads={} channels={}", busy.unread, - busy.attention, + busy.mentions, busy.threads.len(), page.channels.len() ); @@ -91,7 +96,10 @@ async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) { #[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 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()) @@ -107,10 +115,18 @@ async fn bench_unread() { 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, + 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]; @@ -123,7 +139,10 @@ async fn bench_unread() { top(&db, c, *ch, &e).await; firsts.push(e); } - assert_eq!(apply(&db, c, &actor, channel_read(busy, &firsts[0])).await, IntentOutcome::Applied); + 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()); @@ -144,7 +163,10 @@ async fn bench_unread() { } 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); + 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; } @@ -155,7 +177,11 @@ async fn bench_unread() { 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 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); } @@ -193,7 +219,10 @@ async fn bench_unread() { } 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); + 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/context.rs b/crates/buzz-db/src/store/personal_read/context.rs index 028cf78b6a..901e86cb86 100644 --- a/crates/buzz-db/src/store/personal_read/context.rs +++ b/crates/buzz-db/src/store/personal_read/context.rs @@ -50,7 +50,7 @@ impl Db { // The positions the sidebar counts against, both floored at the // account's start and absent before it: the channel's, never // before joining and absent for a non-member; and the thread's, - // which exists only for the actor's threads and includes any + // which exists only for threads the actor follows and includes any // whole-channel cut. let positions = sqlx::query( "SELECT CASE WHEN m.joined_at IS NOT NULL AND s.started IS NOT NULL @@ -65,7 +65,7 @@ impl Db { LEFT JOIN personal_read_frontiers cf ON cf.community_id=$1 AND cf.actor=$2 AND cf.channel_id=$3 AND cf.root_id=''::bytea LEFT JOIN personal_read_frontiers tf ON tf.community_id=$1 AND tf.actor=$2 - AND tf.channel_id=$3 AND tf.root_id=$4 AND $4<>''::bytea", + AND tf.channel_id=$3 AND tf.root_id=$4 AND $4<>''::bytea AND tf.following", ) .bind(community.as_uuid()) .bind(actor_bytes.as_slice()) diff --git a/crates/buzz-db/src/store/personal_read/membership.rs b/crates/buzz-db/src/store/personal_read/membership.rs index 8b2db9fb54..d84461dbc3 100644 --- a/crates/buzz-db/src/store/personal_read/membership.rs +++ b/crates/buzz-db/src/store/personal_read/membership.rs @@ -1,5 +1,7 @@ -//! Thread membership. A thread frontier row exists exactly for the actor's -//! threads, and only ingest creates one: read intents move existing rows. +//! Read state written by ingest: posting marks the author read, and replies +//! and mentions follow threads. A thread frontier row exists for each thread +//! the actor follows or has unfollowed (`following=false`, so unfollow +//! sticks); only ingest and follow intents create one. use super::model::ELIGIBLE_KINDS; use crate::Result; use buzz_core::CommunityId; @@ -8,17 +10,34 @@ use nostr::Event; use sqlx::{Postgres, Transaction}; use uuid::Uuid; -/// Record who a newly stored reply makes a thread member: its author, the -/// root's author, members it mentions, and in a DM (at most nine members) -/// every member. The author's row starts at the reply, which it has read; -/// everyone else's starts just before it, so the reply is unread. Existing -/// rows are left alone. -pub(crate) async fn record_reply( +/// Where a newly stored channel message sits. +#[derive(Clone, Copy)] +pub(crate) enum Place<'a> { + /// A top-level timeline message. + TopLevel, + /// A reply in `root`'s thread; `broadcast` when it also shows on the + /// timeline (a depth-1 reply broadcast to the channel). + Reply { root: &'a [u8], broadcast: bool }, +} + +/// Record what a newly stored message changes: +/// - its author has read through it, in each context it shows in, and a +/// reply (re-)follows the author; +/// - members it mentions follow the thread it is in, or the thread rooted at +/// it; +/// - a reply also makes the root's author, and in a DM (at most nine +/// members) every member, follow the thread. +/// +/// Nobody but the author is re-followed: an unfollowed row stays unfollowed, +/// even for a new mention (whether a mention should re-follow is open). +/// +/// Everyone but the author starts just before the message, so it is unread. +pub(crate) async fn record_message( tx: &mut Transaction<'_, Postgres>, community: CommunityId, event: &Event, channel: Uuid, - root: &[u8], + place: Place<'_>, received_at: DateTime, ) -> Result<()> { if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) { @@ -29,29 +48,54 @@ pub(crate) async fn record_reply( .public_keys() .map(|key| key.to_bytes().to_vec()) .collect(); - // Both inserts' fence and foreign-key checks run at the end of the - // statement, after the account rows exist. + let (root, reply, on_timeline) = match place { + Place::TopLevel => (event.id.as_bytes().as_slice(), false, true), + Place::Reply { root, broadcast } => (root, true, broadcast), + }; + // Rank 0 is the replying author, who follows (again) and has read; 1 a + // mentioned member and 2 anyone else the reply makes follow, whose + // existing rows are left alone. Every insert's fence and foreign-key + // checks run at the end of the statement, after the account rows exist. sqlx::query( "WITH joined AS ( - SELECT $4::bytea AS actor, $6::timestamptz AS through + SELECT $4::bytea AS actor, $6::timestamptz AS through, 0 AS rank WHERE $8 UNION ALL - SELECT pubkey, $6-interval '1 microsecond' FROM events - WHERE community_id=$1 AND channel_id=$2 AND id=$3 + SELECT cm.pubkey, $6-interval '1 microsecond', 1 FROM channel_members cm + WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL + AND cm.pubkey=ANY($5) AND cm.pubkey<>$4 UNION ALL - SELECT cm.pubkey, $6-interval '1 microsecond' + SELECT pubkey, $6-interval '1 microsecond', 2 FROM events + WHERE $8 AND community_id=$1 AND channel_id=$2 AND id=$3 + UNION ALL + SELECT cm.pubkey, $6-interval '1 microsecond', 2 FROM channel_members cm JOIN channels c ON c.community_id=cm.community_id AND c.id=cm.channel_id - WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL - AND (cm.pubkey=ANY($5) OR c.channel_type='dm') + WHERE $8 AND cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL + AND c.channel_type='dm' ), members AS ( - SELECT DISTINCT ON (actor) actor, through FROM joined ORDER BY actor, through DESC + SELECT DISTINCT ON (actor) actor, through, rank FROM joined ORDER BY actor, rank ), accounts AS ( INSERT INTO personal_read_accounts (community_id, actor) - SELECT $1, actor FROM members ON CONFLICT DO NOTHING + SELECT $1, actor FROM members UNION SELECT $1, $4 + ON CONFLICT DO NOTHING + ), timeline AS ( + INSERT INTO personal_read_frontiers + (community_id, actor, channel_id, root_id, through_timestamp) + SELECT $1, $4, $2, ''::bytea, $6 WHERE $7 + ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE + SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, + excluded.through_timestamp) + ), replied AS ( + INSERT INTO personal_read_frontiers + (community_id, actor, channel_id, root_id, through_timestamp) + SELECT $1, actor, $2, $3, through FROM members WHERE rank=0 + ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE + SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, + excluded.through_timestamp), following=true ) INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id, through_timestamp) - SELECT $1, actor, $2, $3, through FROM members + SELECT $1, actor, $2, $3, through FROM members WHERE rank>0 ON CONFLICT DO NOTHING", ) .bind(community.as_uuid()) @@ -60,6 +104,8 @@ pub(crate) async fn record_reply( .bind(event.pubkey.to_bytes().as_slice()) .bind(&mentioned) .bind(received_at) + .bind(on_timeline) + .bind(reply) .execute(&mut **tx) .await?; Ok(()) diff --git a/crates/buzz-db/src/store/personal_read/mod.rs b/crates/buzz-db/src/store/personal_read/mod.rs index 7dfd7e0b93..79657bf565 100644 --- a/crates/buzz-db/src/store/personal_read/mod.rs +++ b/crates/buzz-db/src/store/personal_read/mod.rs @@ -1,8 +1,9 @@ //! Private, signer-owned accessory read progress, separate from NIP-RS events. //! //! A frontier is the relay arrival time of the message a context was read -//! through, and unread counts forward from it. Only fixed context intents -//! advance frontiers; ingest only creates thread membership rows. +//! through, and unread counts forward from it. Fixed intents advance and +//! follow; ingest advances the author's own frontiers (posting marks read) +//! and creates follow rows for replies and mentions. mod classification; mod context; @@ -11,7 +12,7 @@ mod model; mod projection; mod writes; -pub(crate) use membership::record_reply; +pub(crate) use membership::{record_message, Place}; pub use model::*; #[cfg(test)] diff --git a/crates/buzz-db/src/store/personal_read/model.rs b/crates/buzz-db/src/store/personal_read/model.rs index d013c8283b..e06666f505 100644 --- a/crates/buzz-db/src/store/personal_read/model.rs +++ b/crates/buzz-db/src/store/personal_read/model.rs @@ -18,7 +18,8 @@ pub struct ReadTarget { #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] pub enum ReadIntent { - /// Advance a context through one fixed message, including equal arrivals. + /// Advance a context through one fixed message: everything displayed at + /// or before it in that context, and anything with an equal arrival. MarkThrough { /// Channel or canonical thread being marked. target: ReadTarget, @@ -26,13 +27,25 @@ pub enum ReadIntent { message_id: String, }, /// Advance the channel timeline and every thread in it through one fixed - /// message's arrival time. The anchor may be a reply; ancestry is irrelevant. + /// message: everything in the channel displayed at or before it. The + /// anchor may be a reply; ancestry is irrelevant. MarkChannelRead { /// Channel being marked, including all of its threads. channel_id: Uuid, /// Fixed anchor; retry must not substitute the latest message. message_id: String, }, + /// Follow a thread. A thread you did not follow starts caught up. + Follow { + /// The thread; a target without a root is invalid. + target: ReadTarget, + }, + /// Stop following a thread. Sticky: later replies and mentions do not + /// re-follow; your own reply or `follow` does. + Unfollow { + /// The thread; a target without a root is invalid. + target: ReadTarget, + }, } /// Outcome for one independent transaction, never acknowledged before commit. @@ -49,8 +62,6 @@ pub enum IntentOutcome { /// Maximum channel summaries in one sidebar page. pub const MAX_CHANNELS: usize = 20; -/// Maximum unread-thread summaries per channel row. -pub const MAX_THREAD_SUMMARIES: usize = 5; /// 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. @@ -59,6 +70,7 @@ pub const MAX_ARRIVAL_SKEW_SECONDS: u32 = 900; pub const ELIGIBLE_KINDS: [i32; 4] = [9, 40002, 45001, 45003]; /// One joined-channel summary, not a second conversation/history API. +/// Clients receive message IDs only; positions stay relay-internal. #[derive(Debug, Serialize)] pub struct ChannelReadSummary { /// Joined channel UUID. @@ -71,34 +83,33 @@ pub struct ChannelReadSummary { pub archived: bool, /// Existing DM visibility preference (not an authorization decision). pub hidden: bool, - /// An unread timeline message without a [`Reason`] exists: ordinary - /// backlog. Never true in a DM, where every message is direct. + /// The channel timeline has a message past the actor's position. Thread + /// replies never set it; they show on their thread row. 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 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 - /// exactly when it is. - pub latest_message_at: Option, - /// The actor's threads with unread replies, newest unread reply first. + /// Unread timeline messages that mention the actor. Exact, uncapped. + pub mentions: i64, + /// The last timeline message, in display order, at or before the actor's + /// position: the "new" divider goes below it. None when there is none. + pub read_through_id: Option, + /// Followed threads with an unread reply, newest unread reply first. pub threads: Vec, } -/// One of the actor's threads with an unread reply. No conversation bytes. +/// One followed thread with an unread reply. No conversation bytes. #[derive(Debug, Deserialize, Serialize)] pub struct ThreadReadSummary { /// Canonical thread-root event ID. pub root_id: String, - /// 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 - /// necessarily latest_reply_id's own; summaries order by it. - pub latest_reply_at: i64, + /// A reply past the actor's thread position. Listed rows are unread. + pub unread: bool, + /// Unread replies that mention the actor. Exact, uncapped. + pub mentions: i64, + /// The last reply, in display order, at or before the actor's thread + /// position. None when the actor has read no reply. + pub read_through_id: Option, + /// Last unread reply to arrive: marking through it reads the thread, and + /// clients fetch it by ID for a preview. + pub latest_id: String, } /// A bounded roster page, with no cross-page snapshot or removal inference. 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 5c09bb2998..5a0849c274 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -236,7 +236,7 @@ async fn accounts(pool: &PgPool, community: CommunityId) -> i64 { #[tokio::test] #[ignore = "requires Postgres"] -async fn unread_is_presence_and_serializes_as_booleans() { +async fn sidebar_row_has_the_wire_shape() { let (db, _, community, channel, actor, _) = fixture().await; let author = Keys::generate(); let base = now(); @@ -246,18 +246,37 @@ async fn unread_is_presence_and_serializes_as_booleans() { } let row = sidebar(&db, community, &actor).await; let wire = serde_json::to_value(&row).unwrap(); + let mut keys: Vec<_> = wire.as_object().unwrap().keys().cloned().collect(); + keys.sort(); + assert_eq!( + keys, + [ + "archived", + "channel_id", + "channel_type", + "hidden", + "mentions", + "name", + "read_through_id", + "threads", + "unread" + ] + ); 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}"); - } + assert_eq!(wire["mentions"], 0); + assert_eq!( + wire["read_through_id"], + serde_json::Value::Null, + "nothing read yet" + ); let last = last.unwrap(); - assert_eq!(row.latest_message_id, Some(last.id.to_hex())); assert_eq!( apply(&db, community, &actor, mark(channel, None, &last)).await, IntentOutcome::Applied ); - assert!(!sidebar(&db, community, &actor).await.unread); + let row = sidebar(&db, community, &actor).await; + assert!(!row.unread); + assert_eq!(row.read_through_id, Some(last.id.to_hex())); } #[tokio::test] @@ -279,11 +298,11 @@ async fn joining_starts_caught_up_and_counts_only_what_arrives_after() { .unwrap(); let row = sidebar(&db, community, &joiner).await; assert_eq!( - (row.unread, row.attention), - (false, false), + (row.unread, row.mentions), + (false, 0), "history before joining is read" ); - assert_eq!(row.latest_message_id, Some(event.id.to_hex())); + assert_eq!(row.read_through_id, Some(event.id.to_hex())); assert_eq!( status(&db, community, &joiner, channel, None, &[&event]).await, ["read"] @@ -303,7 +322,11 @@ 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), (false, true), "a mention is attention, not backlog"); + assert_eq!( + (row.unread, row.mentions), + (true, 1), + "a mention is unread and counted" + ); assert_eq!( states(&db, community, &joiner, channel, None, &[&later]).await, [serde_json::json!({"status":"unread","reason":"mention"})] @@ -402,11 +425,7 @@ 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.threads.len()), - (false, false, 0) - ); - assert_eq!(row.latest_message_id, Some(before.id.to_hex())); + assert_eq!((row.unread, row.mentions, row.threads.len()), (false, 0, 0)); assert_eq!( status(&db, community, &member, channel, None, &[&before]).await, ["not_counted"] @@ -436,7 +455,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.attention, row.threads.len()), (true, true, 1)); + assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1)); } #[tokio::test] @@ -535,12 +554,12 @@ 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"], true); - assert_eq!(before, after, "not unread, not attention, not latest"); + assert_eq!(before, after, "not unread, not a mention"); } #[tokio::test] #[ignore = "requires Postgres"] -async fn latest_includes_own_and_excludes_deleted_and_auxiliary() { +async fn posting_marks_read_and_read_through_skips_deleted_and_auxiliary() { let (db, pool, community, channel, actor, _) = fixture().await; let base = now(); let own = post(&db, community, channel, &actor, base + 1, vec![]).await; @@ -553,10 +572,13 @@ async fn latest_includes_own_and_excludes_deleted_and_auxiliary() { .await .unwrap(); 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!(row.unread, "own, deleted and reaction do not count"); - // An empty channel has no latest message. + assert!(!row.unread, "posting read through everything before it"); + assert_eq!( + row.read_through_id, + Some(own.id.to_hex()), + "not deleted, not a reaction" + ); + // An empty channel has nothing read. db.create_channel( community, "truly empty", @@ -577,7 +599,7 @@ async fn latest_includes_own_and_excludes_deleted_and_auxiliary() { .iter() .find(|c| c.channel_id != channel) .unwrap(); - assert!(empty.latest_message_id.is_none() && empty.latest_message_at.is_none()); + assert!(!empty.unread && empty.mentions == 0 && empty.read_through_id.is_none()); } #[tokio::test] diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index 78cfc0928d..995bd584a1 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -1,6 +1,8 @@ -//! 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. +//! Sidebar read state from read positions. `unread` is an existence probe +//! that stops at the first qualifying row, and `mentions` walks only the +//! actor's mentions past the position, so a caught-up reader examines almost +//! nothing. Posting marks the author read at ingest, so no probe filters out +//! the actor's own messages. use super::{model::*, writes}; use buzz_core::CommunityId; @@ -17,8 +19,8 @@ const ON_TIMELINE: &str = "(tm.root_event_id IS NULL OR tm.root_event_id=e.id /// 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 UNREAD_EVENT: &str = "e.created_at >= r.position-$7 AND e.received_at > r.position + AND e.kind=ANY($6) 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"; @@ -77,19 +79,18 @@ impl Db { // Every position is floored at the account's start (infinity before // 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. + // up. A thread's is its row, which exists only for threads the actor + // follows (see `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. + // - unread: the first timeline message past the position + // (idx_events_community_channel_created); + // - mentions: the actor's mentions in that scope that arrived past + // the position (idx_event_mentions_scope_received); + // - read_through_id: the newest timeline message, by author time, that + // arrived at or before the position. Anything that arrived by then + // was authored no later than the position plus the skew; + // - threads: followed roots whose last reply arrived after the + // thread position, then their unread replies and mentions. let sql = format!( r#"WITH roster AS MATERIALIZED ( SELECT c.id, c.name, c.channel_type::text AS channel_type, @@ -108,84 +109,64 @@ 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), (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, - ordinary.id IS NOT NULL AS unread, - directed.id IS NOT NULL OR threads.id IS NOT NULL AS attention, + EXISTS (SELECT 1 FROM events e {EVENT_THREAD_ROW} + WHERE e.community_id=$1 AND e.channel_id=r.id + AND {UNREAD_EVENT} AND {ON_TIMELINE}) AS unread, + (SELECT count(*) 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 + WHERE m.community_id=$1 AND m.pubkey_hex=$8 AND m.channel_id=r.id + AND m.root_id IS NULL AND m.received_at > r.position + AND e.channel_id=r.id AND {UNREAD_EVENT}) AS mentions, + (SELECT encode(e.id,'hex') FROM events e {EVENT_THREAD_ROW} + WHERE e.community_id=$1 AND e.channel_id=r.id + AND e.created_at <= r.position+$7 AND e.received_at <= r.position + AND e.kind=ANY($6) AND e.deleted_at IS NULL AND {ON_TIMELINE} + ORDER BY e.created_at DESC, e.id LIMIT 1) AS read_through_id, COALESCE(threads.items, '[]') AS threads FROM roster r - LEFT JOIN LATERAL ( - SELECT (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id, - max(received_at) AS arrival, max(extract(epoch FROM created_at)::bigint) AS at - FROM (SELECT id,created_at,received_at FROM events - WHERE community_id=$1 AND channel_id=r.id AND kind=ANY($6) AND deleted_at IS NULL - ORDER BY created_at DESC LIMIT $7) newest - ) latest ON true - LEFT JOIN LATERAL ( - 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'), - '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 + 'unread',true,'mentions',nm.mentions,'read_through_id',rt.id, + 'latest_id',n.id) ORDER BY n.at DESC, tf.root_id) AS items 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 (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 + max(e.created_at) AS 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 tm.event_created_at >= p.position-$7 + AND e.received_at > p.position AND e.kind=ANY($6) AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast) ) n + CROSS JOIN LATERAL ( + SELECT count(*) AS mentions + 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 + WHERE m.community_id=$1 AND m.pubkey_hex=$8 AND m.channel_id=r.id + AND m.root_id=tf.root_id AND m.received_at > p.position + AND e.kind=ANY($6) AND e.deleted_at IS NULL + ) nm + LEFT JOIN LATERAL ( + SELECT encode(tm.event_id,'hex') AS id + 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+$7 + AND e.received_at <= p.position AND e.kind=ANY($6) AND e.deleted_at IS NULL + ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1 + ) rt ON true WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id - AND tf.root_id<>''::bytea AND root.last_reply_received_at > p.position - AND n.id IS NOT NULL + AND tf.root_id<>''::bytea AND tf.following + AND root.last_reply_received_at > p.position AND n.id IS NOT NULL ) threads ON true ORDER BY r.id"# - ) - .replace("$hex", "$9"); + ); // Only compile-time constants are interpolated. let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(community.as_uuid()) @@ -194,7 +175,6 @@ impl Db { .bind(only) .bind((limit + 1) as i64) .bind(ELIGIBLE_KINDS.as_slice()) - .bind(LATEST_PROBE as i64) .bind(skew()) .bind(actor.to_hex()) .fetch_all(&mut *tx) @@ -203,8 +183,9 @@ 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")?) - .map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?; + let threads: Vec = + serde_json::from_value(row.try_get("threads")?) + .map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?; channels.push(ChannelReadSummary { channel_id: row.try_get("id")?, name: row.try_get("name")?, @@ -212,10 +193,9 @@ impl Db { archived: row.try_get("archived")?, hidden: row.try_get("hidden")?, 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), + mentions: row.try_get("mentions")?, + read_through_id: row.try_get("read_through_id")?, + threads, }); } let next_cursor = if has_more { @@ -230,55 +210,6 @@ impl Db { } } -/// Newest messages, by author time, that join the counted ones as latest candidates. -pub(super) const LATEST_PROBE: usize = 32; - pub(super) fn skew() -> chrono::Duration { chrono::Duration::seconds(i64::from(MAX_ARRIVAL_SKEW_SECONDS)) } - -/// 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| { - b.latest_reply_at - .cmp(&a.latest_reply_at) - .then_with(|| a.root_id.cmp(&b.root_id)) - }); - items.truncate(MAX_THREAD_SUMMARIES); - items -} - -#[cfg(test)] -mod tests { - use super::*; - - fn item(root: u8, at: i64) -> ThreadReadSummary { - ThreadReadSummary { - root_id: hex::encode([root; 32]), - latest_reply_id: hex::encode([root; 32]), - latest_reply_at: at, - } - } - - #[test] - fn thread_summaries_order_newest_first_break_ties_by_root_and_cap_at_five() { - // Literal contract boundaries deliberately do not derive from the constant. - for count in [0_u8, 1, 4, 5, 6, 7] { - // Descending roots with pairwise-equal times exercise the tie-break. - let items: Vec<_> = (0..count) - .rev() - .map(|i| item(i, i64::from(i / 2))) - .collect(); - let summaries = summarize(items); - let roots: Vec<_> = summaries.iter().map(|t| t.root_id.clone()).collect(); - let mut expected: Vec<_> = (0..count).map(|i| (i64::from(i / 2), i)).collect(); - expected.sort_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1))); - let expected: Vec<_> = expected - .into_iter() - .take(5) - .map(|(_, i)| hex::encode([i; 32])) - .collect(); - assert_eq!(roots, expected, "count {count}"); - } - } -} 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 918502424d..10eddf912d 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 @@ -14,39 +14,38 @@ async fn sidebar_sql_eligibility_matches_selector() { message_ids: vec![event.id.to_hex()], }]; for kind in [9, 40002, 45001, 45003, 1, 7, 39002, 40008] { - for own in [false, true] { - for deleted in [false, true] { - sqlx::query("UPDATE events SET kind=$2,pubkey=$3,deleted_at=CASE WHEN $4 THEN now() ELSE NULL END WHERE community_id=$1") - .bind(community.as_uuid()).bind(kind) - .bind(if own { actor.public_key().to_bytes() } else { event.pubkey.to_bytes() }.as_slice()) - .bind(deleted).execute(&pool).await.unwrap(); - let page = db - .personal_read_sidebar(community, &actor.public_key(), 20, None) - .await - .unwrap(); - let expected = u32::from(ELIGIBLE_KINDS.contains(&kind) && !own && !deleted); - assert_eq!( - u32::from(page.channels[0].unread), expected, - "kind={kind} own={own} deleted={deleted}" - ); - let contexts = db - .personal_read_contexts(community, &actor.public_key(), &query) - .await - .unwrap(); - assert_eq!( - serde_json::to_value(&contexts).unwrap()["contexts"][0]["messages"][0] - ["status"], - ["not_counted", "unread"][expected as usize], - "kind={kind} own={own} deleted={deleted}" - ); - } + // Own messages need no rule here: posting marks the author read at + // ingest (`posting_marks_read_and_read_through_skips_deleted_and_auxiliary`). + for deleted in [false, true] { + sqlx::query("UPDATE events SET kind=$2,deleted_at=CASE WHEN $3 THEN now() ELSE NULL END WHERE community_id=$1") + .bind(community.as_uuid()).bind(kind) + .bind(deleted).execute(&pool).await.unwrap(); + let page = db + .personal_read_sidebar(community, &actor.public_key(), 20, None) + .await + .unwrap(); + let expected = u32::from(ELIGIBLE_KINDS.contains(&kind) && !deleted); + assert_eq!( + u32::from(page.channels[0].unread), + expected, + "kind={kind} deleted={deleted}" + ); + let contexts = db + .personal_read_contexts(community, &actor.public_key(), &query) + .await + .unwrap(); + assert_eq!( + serde_json::to_value(&contexts).unwrap()["contexts"][0]["messages"][0]["status"], + ["not_counted", "unread"][expected as usize], + "kind={kind} deleted={deleted}" + ); } } } #[tokio::test] #[ignore = "requires Postgres"] -async fn sidebar_directed_fact_matches_selector_classifier() { +async fn sidebar_mentions_match_selector_classifier() { let (db, pool, community, channel, actor, event) = fixture().await; let actor_hex = actor.public_key().to_hex(); let fullwidth: String = actor_hex @@ -79,9 +78,9 @@ async fn sidebar_directed_fact_matches_selector_classifier() { .await .unwrap(); for tags in &cases { - // A top-level message: `broadcast` tags alone direct nothing. + // A top-level message: `broadcast` tags alone direct nothing, and + // in a DM every message is direct but only mentions count. 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)) @@ -94,13 +93,13 @@ async fn sidebar_directed_fact_matches_selector_classifier() { .execute(&pool) .await .unwrap(); - let mentioned = tags.iter().any(|t| { - t.len() >= 2 && t[0] == "p" && t[1].eq_ignore_ascii_case(&actor_hex) - }); + 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", + "INSERT INTO event_mentions (community_id,pubkey_hex,event_id,event_created_at,channel_id,event_kind,received_at) + SELECT community_id,$2,id,created_at,channel_id,kind,received_at FROM events WHERE community_id=$1 AND id=$3", ) .bind(community.as_uuid()) .bind(&actor_hex) @@ -116,10 +115,13 @@ async fn sidebar_directed_fact_matches_selector_classifier() { .channels .remove(0); assert_eq!( - (row.unread, row.attention), - expected, + (row.unread, row.mentions), + (true, i64::from(mentioned)), "channel_type={channel_type} tags={tags:?}" ); + if channel_type != "dm" { + assert_eq!(reason.is_some(), mentioned, "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 c6fe011cbe..449e8169cd 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,12 +65,11 @@ 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.threads.len()), - (false, true, 1) - ); - assert_eq!(row.threads[0].latest_reply_id, own.id.to_hex()); + // Only the reply is unread: posting read the root, and a reply never + // makes the channel unread. + assert_eq!((row.unread, row.mentions, row.threads.len()), (false, 0, 1)); + assert_eq!(row.threads[0].latest_id, own.id.to_hex()); + assert_eq!(row.threads[0].read_through_id, None, "no reply read yet"); assert_eq!( states(&db, community, &starter, channel, Some(&root), &[&own]).await, [serde_json::json!({"status":"unread","reason":"conversation"})] @@ -92,11 +91,13 @@ async fn replying_mentioning_and_starting_a_thread_make_members() { ) .await; let row = sidebar(&db, community, &actor).await; + assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1)); + assert_eq!(row.threads[0].latest_id, other.id.to_hex()); assert_eq!( - (row.unread, row.attention, row.threads.len()), - (true, true, 1) + row.threads[0].read_through_id, + Some(own.id.to_hex()), + "posting read the actor's own reply" ); - 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. let outsider = Keys::generate(); @@ -116,11 +117,9 @@ async fn replying_mentioning_and_starting_a_thread_make_members() { assert_eq!(thread_rows(&pool, community, &bystander).await, root_row); assert!(thread_rows(&pool, community, &outsider).await.is_empty()); let row = sidebar(&db, community, &bystander).await; - // The mention, and the root on the timeline. - assert_eq!( - (row.unread, row.attention, row.threads.len()), - (true, true, 1) - ); + // The root on the timeline; the mention counts on the thread row only. + assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1)); + assert_eq!(row.threads[0].mentions, 1); assert_eq!( states( &db, @@ -262,10 +261,8 @@ async fn dm_replies_make_every_member_a_thread_member() { .unwrap() .channels .remove(0); - assert_eq!( - (row.unread, row.attention, row.threads.len()), - (false, true, 1) - ); + // The root is unread on the timeline, the reply on the thread row. + assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1)); } #[tokio::test] @@ -316,13 +313,9 @@ 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, 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()); + assert!(row.unread, "top at +20 remains"); + assert_eq!(row.threads.len(), 1, "reply at +30 remains"); + assert_eq!(row.threads[0].latest_id, second.id.to_hex()); assert_eq!( apply(&db, community, &actor, channel_read(channel, &second)).await, @@ -330,8 +323,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.attention, row.threads.len()), - (false, false, 0), + (row.unread, row.threads.len()), + (false, 0), "every thread is covered" ); @@ -359,8 +352,8 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() { ) .await; let row = sidebar(&db, community, &actor).await; - assert_eq!((row.unread, row.attention), (false, true)); - assert_eq!(row.threads[0].latest_reply_id, newer.id.to_hex()); + assert!(!row.unread); + assert_eq!(row.threads[0].latest_id, newer.id.to_hex()); assert_eq!( status( &db, @@ -412,7 +405,7 @@ async fn mark_channel_read_validates_its_anchor() { #[tokio::test] #[ignore = "requires Postgres"] -async fn thread_summaries_order_cap_and_count_every_unread_thread() { +async fn thread_summaries_order_newest_first_and_list_every_unread_thread() { let (db, _, community, channel, actor, _) = fixture().await; let base = now(); let mut roots = Vec::new(); @@ -456,8 +449,9 @@ 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, row.threads.len()), (true, true, 5)); + // Posting the roots read the timeline through them; no thread cap. + assert_eq!((row.unread, row.threads.len()), (false, 6)); + assert_eq!(row.read_through_id, Some(roots[5].id.to_hex())); 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(); @@ -468,22 +462,21 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() { tied[1].clone(), roots[2].id.to_hex(), roots[3].id.to_hex(), - roots[4].id.to_hex() + roots[4].id.to_hex(), + roots[5].id.to_hex() ] ); - 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. - assert_eq!(row.latest_message_at, Some((base + 150) as i64)); + // Ordered by its newest reply's author time, anchored at its last arrival. + assert_eq!(row.threads[2].latest_id, last.unwrap().id.to_hex()); - // Reading a listed thread through its anchor promotes the sixth. + // Reading a listed thread through its anchor removes only it. let first = &row.threads[0]; let intent = ReadIntent::MarkThrough { target: ReadTarget { channel_id: channel, root_id: Some(first.root_id.clone()), }, - message_id: first.latest_reply_id.clone(), + message_id: first.latest_id.clone(), }; assert_eq!( apply(&db, community, &actor, intent).await, @@ -492,8 +485,7 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() { let row = sidebar(&db, community, &actor).await; 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!(row.unread); + assert!(!row.unread); } #[tokio::test] @@ -536,11 +528,13 @@ 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, row.attention), (true, true), "the root and two replies"); + // The fixture's root; each reply's mention counts on its own thread. + assert_eq!((row.unread, row.mentions), (true, 0)); assert_eq!(row.threads.len(), 2); + assert!(row.threads.iter().all(|t| t.mentions == 1)); let listed = &row.threads[1]; assert_eq!(listed.root_id, diff.id.to_hex()); - assert_eq!(listed.latest_reply_id, on_diff.id.to_hex()); + assert_eq!(listed.latest_id, on_diff.id.to_hex()); assert_eq!( states(&db, community, &actor, channel, Some(&diff), &[&on_diff]).await, [serde_json::json!({"status":"unread","reason":"mention"})] @@ -559,9 +553,9 @@ 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, row.attention), (true, true)); + assert!(row.unread); assert_eq!(row.threads.len(), 1); - assert_eq!(row.threads[0].latest_reply_id, elsewhere.id.to_hex()); + assert_eq!(row.threads[0].latest_id, elsewhere.id.to_hex()); } #[tokio::test] @@ -608,3 +602,177 @@ async fn targeted_sidebar_returns_only_requested_joined_channels() { .is_err()); } } + +fn follow(channel: Uuid, root: &nostr::Event, following: bool) -> ReadIntent { + let target = ReadTarget { + channel_id: channel, + root_id: Some(root.id.to_hex()), + }; + if following { + ReadIntent::Follow { target } + } else { + ReadIntent::Unfollow { target } + } +} + +async fn following(pool: &PgPool, community: CommunityId, actor: &Keys) -> Vec { + sqlx::query_scalar( + "SELECT following FROM personal_read_frontiers + WHERE community_id=$1 AND actor=$2 AND root_id<>''::bytea", + ) + .bind(community.as_uuid()) + .bind(actor.public_key().to_bytes().as_slice()) + .fetch_all(pool) + .await + .unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn follow_starts_caught_up_and_unfollow_is_sticky() { + let (db, pool, community, channel, actor, root) = fixture().await; + let base = now(); + let other = Keys::generate(); + let say = |at: u64, tags: Vec| { + let (db, root, other) = (db.clone(), root.clone(), other.clone()); + async move { reply(&db, community, channel, &root, &other, at, tags, false).await } + }; + say(base, vec![]).await; + assert!(following(&pool, community, &actor).await.is_empty()); + + // Follow starts at the thread's last reply: nothing unread yet. + assert_eq!( + apply(&db, community, &actor, follow(channel, &root, true)).await, + IntentOutcome::Applied + ); + assert!(sidebar(&db, community, &actor).await.threads.is_empty()); + let next = say(base + 1, vec![]).await; + let row = sidebar(&db, community, &actor).await; + assert_eq!(row.threads.len(), 1); + assert_eq!(row.threads[0].latest_id, next.id.to_hex()); + + // Unfollow keeps the row; later replies, mentions included, don't + // re-follow (whether a new mention should is an open product question). + assert_eq!( + apply(&db, community, &actor, follow(channel, &root, false)).await, + IntentOutcome::Applied + ); + assert!(sidebar(&db, community, &actor).await.threads.is_empty()); + say(base + 2, vec![]).await; + say(base + 3, mention(&actor)).await; + assert_eq!(following(&pool, community, &actor).await, [false]); + assert!(sidebar(&db, community, &actor).await.threads.is_empty()); + + // The actor's own reply re-follows, read through it. + let own = reply( + &db, + community, + channel, + &root, + &actor, + base + 4, + vec![], + false, + ) + .await; + assert_eq!(following(&pool, community, &actor).await, [true]); + assert!(sidebar(&db, community, &actor).await.threads.is_empty()); + say(base + 5, vec![]).await; + let row = sidebar(&db, community, &actor).await; + assert_eq!(row.threads.len(), 1); + assert_eq!( + ( + row.threads[0].mentions, + row.threads[0].read_through_id.clone() + ), + (0, Some(own.id.to_hex())), + "the earlier mention was read when the actor replied" + ); + + // Following again catches up; a target without a root is invalid. + assert_eq!( + apply(&db, community, &actor, follow(channel, &root, false)).await, + IntentOutcome::Applied + ); + assert_eq!( + apply(&db, community, &actor, follow(channel, &root, true)).await, + IntentOutcome::Applied + ); + assert!(sidebar(&db, community, &actor).await.threads.is_empty()); + let channel_only = ReadIntent::Follow { + target: ReadTarget { + channel_id: channel, + root_id: None, + }, + }; + assert_eq!( + apply(&db, community, &actor, channel_only).await, + IntentOutcome::Invalid + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn mentions_count_exactly_in_their_own_scope() { + let (db, _, community, channel, actor, root) = fixture().await; + let base = now(); + let author = Keys::generate(); + let first = post(&db, community, channel, &author, base + 1, mention(&actor)).await; + post(&db, community, channel, &author, base + 2, mention(&actor)).await; + post(&db, community, channel, &author, base + 3, vec![]).await; + for at in [base + 4, base + 5, base + 6] { + reply( + &db, + community, + channel, + &root, + &author, + at, + mention(&actor), + false, + ) + .await; + } + // A broadcast reply shows on the timeline, so its mention counts there. + reply( + &db, + community, + channel, + &root, + &author, + base + 7, + mention(&actor), + true, + ) + .await; + + let row = sidebar(&db, community, &actor).await; + assert_eq!((row.unread, row.mentions), (true, 3)); + let thread = row + .threads + .iter() + .find(|t| t.root_id == root.id.to_hex()) + .unwrap(); + assert_eq!( + thread.mentions, 3, + "thread mentions never count on the channel" + ); + + // Reading through the first mention leaves the rest, still exact. + assert_eq!( + apply(&db, community, &actor, mark(channel, None, &first)).await, + IntentOutcome::Applied + ); + let row = sidebar(&db, community, &actor).await; + assert_eq!(row.mentions, 2); + assert_eq!(row.read_through_id, Some(first.id.to_hex())); + let thread = row + .threads + .iter() + .find(|t| t.root_id == root.id.to_hex()) + .unwrap(); + assert_eq!( + thread.mentions, 3, + "a channel mark leaves thread positions alone" + ); +} diff --git a/crates/buzz-db/src/store/personal_read/writes.rs b/crates/buzz-db/src/store/personal_read/writes.rs index a4940cb5fb..0fba37399d 100644 --- a/crates/buzz-db/src/store/personal_read/writes.rs +++ b/crates/buzz-db/src/store/personal_read/writes.rs @@ -86,6 +86,7 @@ async fn access( struct Message { id: Vec, + created_at: DateTime, received_at: DateTime, /// Canonical thread root, when the message is a reply. thread: Option>, @@ -103,7 +104,7 @@ async fn message( kinds: Option<&[i32]>, ) -> Result> { let row = sqlx::query( - "SELECT e.id, e.received_at, tm.root_event_id, tm.depth, tm.broadcast + "SELECT e.id, e.created_at, e.received_at, tm.root_event_id, tm.depth, tm.broadcast FROM events e LEFT JOIN thread_metadata tm ON tm.community_id=e.community_id AND tm.event_created_at=e.created_at AND tm.event_id=e.id AND tm.channel_id=e.channel_id WHERE e.community_id=$1 AND e.channel_id=$2 AND e.id=$3 @@ -118,6 +119,7 @@ async fn message( && row.try_get::, _>("broadcast")? == Some(true); Ok(Message { on_timeline: thread.is_none() || broadcast_reply, + created_at: row.try_get("created_at")?, received_at: row.try_get("received_at")?, thread, id, @@ -126,15 +128,16 @@ async fn message( .transpose() } -/// An eligible-kind message's arrival time, deleted or not, without tags. -async fn anchor_received_at( +/// An eligible-kind message's author and arrival times, deleted or not, +/// without tags. +async fn anchor_times( conn: &mut PgConnection, community: CommunityId, channel: Uuid, id: &[u8], -) -> Result>> { - Ok(sqlx::query_scalar( - "SELECT received_at FROM events +) -> Result, DateTime)>> { + Ok(sqlx::query_as( + "SELECT created_at, received_at FROM events WHERE community_id=$1 AND channel_id=$2 AND id=$3 AND kind=ANY($4) LIMIT 1", ) .bind(community.as_uuid()) @@ -145,6 +148,47 @@ async fn anchor_received_at( .await?) } +/// Where a mark through an anchor leaves the frontier: the last arrival among +/// the messages displayed (by author time, then ID) at or before the anchor +/// in `scope`. Clients name only display-order IDs, so a late arrival shown +/// above the anchor is read with it instead of leaving a badge no mark can +/// clear. Anything that arrived after the anchor was authored no earlier +/// than the skew before the anchor's arrival, which bounds the range. +/// `scope` is a thread root, empty for the channel timeline, or None for +/// the whole channel. +async fn display_through( + conn: &mut PgConnection, + community: CommunityId, + channel: Uuid, + anchor: (&[u8], DateTime, DateTime), + scope: Option<&[u8]>, +) -> Result> { + let (id, created_at, received_at) = anchor; + let through: Option> = sqlx::query_scalar( + "SELECT max(e.received_at) 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=$2 AND e.kind=ANY($3) + AND e.created_at >= $4::timestamptz-$7 AND e.created_at <= $5 + AND (e.created_at, e.id) <= ($5, $6) AND e.received_at > $4 + AND CASE WHEN $8::bytea IS NULL THEN true + WHEN $8=''::bytea THEN tm.root_event_id IS NULL OR tm.root_event_id=e.id + OR (tm.depth=1 AND tm.broadcast) + ELSE e.id=$8 OR tm.root_event_id=$8 END", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(ELIGIBLE_KINDS.as_slice()) + .bind(received_at) + .bind(created_at) + .bind(id) + .bind(super::projection::skew()) + .bind(scope) + .fetch_one(conn) + .await?; + Ok(through.map_or(received_at, |later| later.max(received_at))) +} + pub(super) async fn valid_target( conn: &mut PgConnection, community: CommunityId, @@ -203,6 +247,14 @@ pub(super) async fn apply( if !msg.on_timeline { return Ok(IntentOutcome::Blocked); } + let through = display_through( + conn, + community, + target.channel_id, + (msg.id.as_slice(), msg.created_at, msg.received_at), + Some(root.as_slice()), + ) + .await?; sqlx::query( "INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id, through_timestamp) @@ -213,13 +265,21 @@ pub(super) async fn apply( .bind(community.as_uuid()) .bind(actor) .bind(target.channel_id) - .bind(msg.received_at) + .bind(through) .execute(&mut *conn) .await?; } else { if msg.id != root && msg.thread.as_ref() != Some(&root) { return Ok(IntentOutcome::Blocked); } + let through = display_through( + conn, + community, + target.channel_id, + (msg.id.as_slice(), msg.created_at, msg.received_at), + Some(root.as_slice()), + ) + .await?; // Reading a thread never joins it: only the actor's threads // have a row to advance. sqlx::query( @@ -231,7 +291,7 @@ pub(super) async fn apply( .bind(actor) .bind(target.channel_id) .bind(&root) - .bind(msg.received_at) + .bind(through) .execute(&mut *conn) .await?; } @@ -246,11 +306,21 @@ pub(super) async fn apply( if !access(conn, community, actor, *channel_id).await? { return Ok(IntentOutcome::Blocked); } - // Only the anchor's arrival matters: ancestry cannot change which - // messages a whole-channel cut covers. - let Some(through) = anchor_received_at(conn, community, *channel_id, &id).await? else { + // Ancestry cannot change which messages a whole-channel cut + // covers: everything displayed at or before the anchor. + let Some((created_at, received_at)) = + anchor_times(conn, community, *channel_id, &id).await? + else { return Ok(IntentOutcome::Blocked); }; + let through = display_through( + conn, + community, + *channel_id, + (&id, created_at, received_at), + None, + ) + .await?; sqlx::query( "INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id, through_timestamp, threads_through_timestamp) VALUES ($1,$2,$3,''::bytea,$4,$4) @@ -260,6 +330,42 @@ pub(super) async fn apply( ).bind(community.as_uuid()).bind(actor).bind(channel_id).bind(through) .execute(&mut *conn).await?; } + ReadIntent::Follow { target } | ReadIntent::Unfollow { target } => { + if target.root_id.is_none() { + return Ok(IntentOutcome::Invalid); + } + let Some(root) = valid_target(conn, community, actor, target).await? else { + return Ok(IntentOutcome::Blocked); + }; + let following = matches!(intent, ReadIntent::Follow { .. }); + // A new row starts caught up: at the thread's last reply, or the + // root itself before any reply. Following again also catches up, + // so replies that arrived while unfollowed stay muted. Unfollow + // keeps the row and its position, so later replies don't + // re-follow. + sqlx::query( + "INSERT INTO personal_read_frontiers + (community_id, actor, channel_id, root_id, through_timestamp, following) + SELECT $1, $2, $3, $4, COALESCE( + (SELECT max(last_reply_received_at) FROM thread_metadata + WHERE community_id=$1 AND channel_id=$3 AND event_id=$4 AND depth=0), + (SELECT max(received_at) FROM events + WHERE community_id=$1 AND channel_id=$3 AND id=$4)), $5 + ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE + SET following=excluded.following, + through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, + excluded.through_timestamp) + WHERE excluded.following AND NOT personal_read_frontiers.following + OR NOT excluded.following", + ) + .bind(community.as_uuid()) + .bind(actor) + .bind(target.channel_id) + .bind(&root) + .bind(following) + .execute(&mut *conn) + .await?; + } } Ok(IntentOutcome::Applied) } 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 07121c4569..4188f28f3c 100644 --- a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs +++ b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs @@ -412,12 +412,15 @@ async fn accessory_write_revocation_is_terminal_before_persistence() { // Exercise the same per-item function the batch handler uses after admission. let result = super::handlers::write_intent(&state, &headers, &principal, &intent).await; assert_eq!(result, json!({"status":"blocked"})); - let persisted: i64 = - sqlx::query_scalar("SELECT count(*) FROM personal_read_accounts WHERE community_id=$1") - .bind(community.as_uuid()) - .fetch_one(state.db.pool()) - .await - .unwrap(); + // Ingest gives the message's author an account; the actor gets none. + let persisted: i64 = sqlx::query_scalar( + "SELECT count(*) FROM personal_read_accounts WHERE community_id=$1 AND actor=$2", + ) + .bind(community.as_uuid()) + .bind(actor.public_key().to_bytes().as_slice()) + .fetch_one(state.db.pool()) + .await + .unwrap(); assert_eq!(persisted, 0, "denied intent must not persist"); } @@ -473,7 +476,6 @@ async fn accessory_discovery_is_host_bound_and_opt_in() { d["max_context_messages"], buzz_db::personal_read::MAX_CONTEXT_MESSAGES ); - assert_eq!(d["max_thread_summaries"], 5); assert_eq!(d["eligible_kinds"], json!([9, 40002, 45001, 45003])); } } @@ -636,7 +638,7 @@ async fn signed_sidebar_deletion(deletion_kind: u16) { .sign_with_keys(signer) .unwrap(); let path = format!("/buzz/v1/me/sidebar?channel_ids={channel}"); - for (event, key, count) in [(&message, &author, 1), (&deletion, signer, 0)] { + for (event, key, unread) in [(&message, &author, true), (&deletion, signer, false)] { let body = serde_json::to_vec(event).unwrap(); let auth = proof(key, &host, "/events", "POST", Some(&body)); // /events has a different response contract from the accessory helper. @@ -669,7 +671,7 @@ async fn signed_sidebar_deletion(deletion_kind: u16) { assert_eq!(channels[0]["channel_id"], channel.to_string()); assert_eq!( channels[0]["unread"], - json!(count), + json!(unread), "kind {deletion_kind}, after kind {}: {sidebar}", event.kind.as_u16() ); @@ -678,12 +680,12 @@ async fn signed_sidebar_deletion(deletion_kind: u16) { #[tokio::test] #[ignore = "requires Postgres"] -async fn accessory_signed_kind5_deletion_clears_another_readers_sidebar_count() { +async fn accessory_signed_kind5_deletion_clears_another_readers_sidebar_unread() { signed_sidebar_deletion(5).await; } #[tokio::test] #[ignore = "requires Postgres"] -async fn accessory_signed_kind9005_deletion_clears_another_readers_sidebar_count() { +async fn accessory_signed_kind9005_deletion_clears_another_readers_sidebar_unread() { signed_sidebar_deletion(9005).await; } diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index d0aa81f54a..70517a803c 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -96,10 +96,8 @@ pub struct BuzzV1Descriptor { pub max_contexts: usize, /// Maximum message selectors across one context request. pub max_context_messages: usize, - /// Maximum unread-thread summaries per sidebar channel row. - pub max_thread_summaries: usize, - /// Message kinds that count as unread and as latest activity. Clients - /// classify live arrivals with this set instead of keeping a copy. + /// Message kinds that count as unread. Clients classify live arrivals + /// with this set instead of keeping a copy. pub eligible_kinds: [i32; 4], } @@ -389,7 +387,6 @@ 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, }; info.buzz_v1 = Some(BuzzV1Descriptor { version: 1, @@ -398,7 +395,6 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st max_intents: MAX_INTENTS, max_contexts: MAX_CONTEXTS, max_context_messages: MAX_CONTEXT_MESSAGES, - max_thread_summaries: MAX_THREAD_SUMMARIES, eligible_kinds: ELIGIBLE_KINDS, }); } diff --git a/docs/buzz-v1-read-state.md b/docs/buzz-v1-read-state.md index b6210a75b4..8cc8a9c1b4 100644 --- a/docs/buzz-v1-read-state.md +++ b/docs/buzz-v1-read-state.md @@ -11,9 +11,8 @@ On a known community host, NIP-11 (`GET /` with `Accept: application/nostr+json`, or `GET /info`) includes `buzz_v1` only when enabled: ```json -{"buzz_v1":{"version":1,"base_path":"/buzz/v1","retention_seconds":2592000, -"max_channels":20,"max_intents":100,"max_contexts":20,"max_context_messages":100, -"max_thread_summaries":5,"eligible_kinds":[9,40002,45001,45003]}} +{"buzz_v1":{"version":1,"base_path":"/buzz/v1","max_channels":20,"max_intents":100, +"max_contexts":20,"max_context_messages":100,"eligible_kinds":[9,40002,45001,45003]}} ``` Use the requesting origin plus this relative prefix. Discovery is a configured @@ -50,16 +49,10 @@ Unknown request fields are rejected. Never turn a transport failure into read. ## Sidebar `GET /buzz/v1/me/sidebar?limit=20&cursor=` returns -`account`, `channels`, and `next_cursor`. Omit the cursor on the first request. -Each channel includes identity/name/type, archived and hidden flags, `unread`, -`attention`, `latest_message_id` (the last eligible message to arrive: the -row's read anchor, see [Order](#order)), `latest_message_at` (the greatest -author time among eligible messages, in seconds: display activity, not -necessarily that message's own time; null exactly when the ID is null), -`latest_message_complete`, and `threads`. Only joined, nondeleted channels are -listed. Hidden/archived presentation remains client-owned. Each page has a -writer-consistent snapshot; separate pages do not share a snapshot, and an -unfinished traversal cannot prove channel removal. +`channels` and `next_cursor`. Omit the cursor on the first request. Only joined, +nondeleted channels are listed. Hidden/archived presentation remains +client-owned. Each page has a writer-consistent snapshot; separate pages do not +share a snapshot, and an unfinished traversal cannot prove channel removal. `GET /buzz/v1/me/sidebar?channel_ids=,` refreshes 1–20 unique channels in one snapshot, ordered by ID with `next_cursor: null`. It cannot be @@ -67,78 +60,61 @@ combined with `limit` or `cursor`. A requested ID absent from the result was not a joined, nondeleted, accessible sidebar row at that snapshot: remove its row. Absence says nothing else about access to an open channel. -`threads` lists unread threads in the row, newest unread reply first: +Each row is one read scope for the channel timeline plus one per followed +thread with an unread reply: ```json -{"items":[{"root_id":"<64-hex>","unread":{"status":"exact","value":2}, - "latest_reply_id":"<64-hex>","latest_reply_at":1700000000}],"complete":true} +{"channel_id":"","name":"general","channel_type":"stream","archived":false, + "hidden":false,"unread":true,"mentions":2,"read_through_id":"<64-hex>", + "threads":[{"root_id":"<64-hex>","unread":true,"mentions":1, + "read_through_id":"<64-hex>","latest_id":"<64-hex>"}]} ``` -Items are canonical roots with unread replies that count (see below), ordered -by `latest_reply_at` descending, then `root_id`; at most 5. `latest_reply_id` is -the last such reply to arrive (equal arrivals prefer the smaller ID), so a -thread `mark_through` at it reads every reply listed; `latest_reply_at` is its -author time. Replies that do not count are filtered out before the count, the -latest reply and the cap are chosen. `unread` uses the row's -definition; every counted reply is also attention, so items carry no separate -attention count. `complete=true` means the unread window was exhausted, no -evidence had unresolved ancestry, unusable tags or undecided membership, and no -thread was omitted; then item unread counts sum to the row's unread replies. -Otherwise the list is a cut of observed evidence and counts may be lower bounds -or unknown. No message bytes are included. +- `unread`: a message in that scope arrived after your read position. The + channel scope is the timeline only (top-level messages and broadcast + replies). A reply in a thread never makes the channel unread; it shows on + its thread row. +- `mentions`: exact, uncapped count of unread messages in that scope that tag + you with `p`. A thread mention counts on the thread row, not the channel. +- `read_through_id`: the last message in that scope, in display order, at or + before your read position, or null when there is none. Draw the "new" + divider below it and open the view there. +- `latest_id` (threads only): the last unread reply to arrive. Fetch it by ID + for a preview; marking the thread through it reads every listed reply. -Counts have exactly three representations: +`threads` lists every followed thread with an unread reply, newest reply +(author time) first, then by `root_id`. There is no cap. A badge or a threads +page is computed on the client from these rows. Clients receive message IDs +only; positions and timestamps stay relay-internal. No message bytes are +included. -```json -[{"status":"exact","value":0},{"status":"at_least","value":7},{"status":"unknown"}] -``` +A message counts when it is nondeleted and of the advertised `eligible_kinds`, +so an edit, reaction or diff (40008) never makes a scope unread. Classify live +arrivals with the advertised set, not a client copy. Your own messages need +no rule: posting marks you read through what you post (see +[Ingest](#ingest-posting-and-following)). -Only exact zero proves absence. Unknown has no numeric value. A message is -eligible when it is non-own, nondeleted, of the advertised `eligible_kinds` and -inside the horizon. The same kinds alone define latest activity, so an edit, -reaction or diff (40008) neither makes a channel unread nor moves it. Classify -live arrivals with the advertised set, not a client copy. +## Ingest: posting and following -An eligible message beyond the matching context frontier counts as unread for -the first reason that holds: +The relay updates read state when it stores an eligible message: -| `reason` | Holds when | -|---|---| -| `direct` | its channel is a DM | -| `mention` | it tags the actor with `p` | -| `conversation` | it is a reply, and the actor wrote its direct parent or has a reply to that same parent, in that channel | -| `broadcast` | it carries `broadcast=1` | -| null | it is top-level | +- **Posting marks read.** The author's position moves through their own + message in each scope it shows in: the timeline for a top-level message or a + broadcast reply, the thread for a reply. +- **Following is a relay-side thread row.** You follow a thread when you reply + to it, when someone replies to your top-level message, when a message + mentions you (the thread it is in, or the thread rooted at it), in a DM when + any member replies, or with a `follow` intent. A new follow starts just + before the message that caused it, so that message is unread. +- **Unfollow is sticky.** `unfollow` keeps the row with `following=false`, so + later replies and mentions do not follow you again. Your own reply, or + `follow`, does. Whether a new mention should re-follow is an open product + question; today it does not. -A reply with no reason does not count: it is not unread and appears in no -thread list. `unread` counts the messages that count; `attention` is the subset -with a reason. Conversation membership uses only the actor's live eligible -messages (a deleted parent proves nothing; a surviving reply to it still does), -looks at the direct parent only (owning the root or replying elsewhere in the -thread proves nothing), and is independent of read progress and retention. A -reply whose membership is undecided is left out, and the row's counts become -lower bounds or unknown. The sidebar counts a broadcast reply without asking -which of the two reasons applies. This is not Desktop notification policy: -follows and mutes do not affect these counts. - -The unread horizon defaults to 30 days (`BUZZ_V1_RETENTION_SECONDS`) and is -measured in author time (`created_at`): a message counts while its author time -is at or after `account.cutoff_ms`. It filters unread/attention, not latest -activity, event storage or frontier state. Frontiers use a different clock, -relay arrival (see [Order](#order)). Three consequences: - -- A message accepted late with an author time beyond the horizon (an import, a - backfill, a long-offline sender) is excluded under the current horizon, - however recently the relay accepted it. It can still be latest when the - horizon holds no message. -- Unread expires at author time plus the horizon, so a future-dated author - time extends how long a message counts. -- A context's unread set is two tests, not one range: arrived after the - frontier, and author time at or after the cutoff. - -A later configuration expansion can change counts without having lost progress. -Latest activity is independent of actor and frontiers. A null latest ID proves -an empty eligible history only when `latest_message_complete=true`. +An account starts caught up at its first read intent (`started_at`): every +position is floored there, so history from before then never counts, and +nothing counts before the first intent. Joining a channel starts that channel +caught up. ## Explicit contexts @@ -157,8 +133,8 @@ and one `contexts` entry per request entry, in order. Context status is frontier includes any whole-channel cut. Message status is `read`, `not_counted`, `unread` (with `reason`), `unknown`, or `unavailable`. Wrong-context, missing and forbidden selectors share unavailable. Status is decided in this order: -ancestry and context; eligibility (`not_counted` for own, deleted, other kinds -and outside the horizon); the frontier (`read`, with no membership lookup); then +ancestry and context; eligibility (`not_counted` for own, deleted, other kinds, +or before the account started); the frontier (`read`, with no membership lookup); then the reason. A reply past the frontier with no reason is `not_counted` when it is proven outside the actor's conversations and `unknown` when membership is undecided. A broadcast reply whose membership is undecided is `unread` with @@ -173,7 +149,9 @@ Conversation bytes must still come from the existing Nostr path. {"intents":[ {"type":"mark_through","target":{"channel_id":""},"message_id":"<64-hex-event>"}, {"type":"mark_through","target":{"channel_id":"","root_id":"<64-hex-root>"},"message_id":"<64-hex-event>"}, - {"type":"mark_channel_read","channel_id":"","message_id":"<64-hex-event>"} + {"type":"mark_channel_read","channel_id":"","message_id":"<64-hex-event>"}, + {"type":"follow","target":{"channel_id":"","root_id":"<64-hex-root>"}}, + {"type":"unfollow","target":{"channel_id":"","root_id":"<64-hex-root>"}} ]} ``` @@ -184,89 +162,85 @@ later failures. `projection_status` is `not_requested`; a successful write does not assert a client has refreshed. Retry the same operands, never substitute latest. Keep pending intent durably on the client until its outcome is resolved. -A mark-through validates a fixed message and advances its context's monotone -frontier to that message's relay arrival (see [Order](#order)). Channel and -thread frontiers never inherit in either direction. Opening a view is not -itself a reading action; client dwell/focus policy determines when to send an -actual observed anchor. Old or deleted valid anchors may advance a frontier. +A mark-through validates a fixed message and reads everything displayed at or +before it in its scope (see [Order](#order)). Channel and thread positions never +inherit in either direction. Opening a view is not itself a reading action; +client dwell/focus policy determines when to send an actual observed anchor. +Old or deleted valid anchors may advance a position. Marking a thread you do +not follow changes nothing. -`mark_channel_read` is the one whole-channel cut: it advances the channel -timeline and every thread in that channel, including unlisted ones, through the -anchor's arrival. The anchor must be an accessible eligible-kind message in -the channel, top-level or reply, deleted or not; ancestry is not checked. A -reply is read at or below the greater of its thread frontier and this cut. An -anchor that no longer exists is `blocked`. `latest_message_id` is the anchor -that reads the whole row. A null ID with `latest_message_complete=false` does -not prove empty history; it only leaves the client without an anchor. Thread -marks and channel `mark_through` never set the cut. +`mark_channel_read` is the one whole-channel cut: it reads the channel +timeline and every thread in that channel, including unlisted ones, through +everything in the channel displayed at or before the anchor. The anchor must be +an accessible eligible-kind message in the channel, top-level or reply, deleted +or not; ancestry is not checked. A reply is read at or below the greater of its +thread position and this cut. An anchor that no longer exists is `blocked`. +Thread marks and channel `mark_through` never set the cut. + +`follow` and `unfollow` take a thread target; one without `root_id` is +`invalid`. Following starts caught up at the thread's last reply, including +when you follow again after unfollowing. ### Order -A frontier is the relay arrival time (`events.received_at`) of the message a -context was read through, never its author time, which the sender chooses and -the relay accepts up to 15 minutes either way. Everything that arrived at or -before the anchor is read, whatever its author time. A message that arrives -later is unread even when backdated, and a future-dated anchor reads nothing -that arrives after it. +A position is a relay arrival time (`events.received_at`), never author time, +which the sender chooses and the relay accepts up to 15 minutes either way. A +message that arrives after your position is unread even when backdated, and a +future-dated message cannot read anything that arrives after it. -The order is the relay's and is not exposed: no response carries a frontier, -and no field lets a client compute what a mark will cover. Send the anchors the -user actually saw, let the relay take the greatest, and ask a context which -messages are read. Do not compare author times, IDs or local receipt order to -drop one pending anchor in favor of another. +Clients see display order (author time, then ID) and name only message IDs. A +mark therefore reads everything displayed at or before its anchor in that +scope: the position becomes the last arrival among them. A late arrival shown +above the newest message is read when you mark that message, so marking the +bottom of a view always clears its badge. The cost, only when a late arrival +sits above the anchor, is that a message displayed below the anchor but arrived +before the late one is read too. With no late arrival, a mark reads exactly +what arrived by the anchor, and a message shown below it stays unread. + +Positions are never exposed. `read_through_id` is the last message displayed at +or before the position; in the rare late-arrival case an unread message can sit +above that divider, while the row's `unread` stays correct. Arrival is the accepting relay process's clock, read just before the insert, at -microsecond resolution. Three limits follow, none of which strands a badge: +microsecond resolution. It is not commit order, relay processes with different +clocks can stamp out of true order by their skew, and messages with the +identical stamp are read together. -- It is not commit order. An insert that commits after a later-stamped message - was already marked read lands read. -- Relay processes with different clocks can stamp out of true order by their - skew. -- Messages with the identical stamp are read together. - -`latest_message_id` is the last message to arrive among those the unread count -examined: the 4,096 most recent events by author time inside the horizon. So -marking through it reads everything counted. When the horizon holds no message, -it is the last to arrive among the channel's 256 most recent events. - -There is no import of earlier client read state: an account starts with no -frontiers, and the horizon bounds what that can show as unread. Manual unread -remains device-local. +There is no import of earlier client read state. Manual unread remains +device-local. ## Bounds and deployment - 20 sidebar rows, 100 intents, 20 contexts / 100 selectors per request. - 64 KiB write body; 16 KiB context URL; 1 MiB serialized API response. -- 4096 raw events per channel inside the horizon plus one exhaustion sentinel, - before eligibility. -- Latest activity probes 256 events plus a sentinel; long ineligible tails may - leave latest incomplete even when unread is exact. -- Tag documents over 8192 bytes or malformed relevant tags yield uncertainty. - Compact boolean facts cross the database boundary, never raw tag payloads. -- Conversation membership: at most 1024 unique parents per request, with a - 500 ms savepoint budget. The lookup is exact, so its work grows with the - replies under each parent. Past either bound a reply is undecided, never - absent: counts become lower bounds or unknown and a context reports `unknown`. +- `unread` is an existence probe that stops at the first message past the + position, so a caught-up reader examines almost nothing. Counting is forward + from the position: work grows with what is unread, not with history. +- Mention counts walk only your mentions in that scope past the position + (`idx_event_mentions_scope_received`). +- A followed thread with nothing new is skipped by comparing the root's + `thread_metadata.last_reply_received_at` with the thread position, without + reading replies. - DB statement/lock deadlines and HTTP read deadlines bound work; writes use a shared eight-second intent-processing deadline after admission. Limits are containment, not a production capacity claim. -Apply migration 0056 (or the equivalent desired schema). It creates two empty -private tables and no index on `events`: both sidebar scans are served by the -existing `idx_events_community_channel_created`. No new per-message ingest write -path or stored unread counters are introduced. +Apply migration 0056 (or the equivalent desired schema). It creates two +private tables, adds `thread_metadata.last_reply_received_at` and +`event_mentions.received_at`/`root_id` (NULL for rows from before it, which no +position can reach), and one index on `event_mentions`. Ingest writes the +author's own position and follow rows in the same transaction as the message. Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and -`/buzz/v1/me/read-state`, plus database pool/statement metrics. Inspect exact / -lower-bound / unknown proportions in controlled acceptance captures; no payload, -actor, channel or frontier values should become metric labels. The measured -local seed is not a DAU/concurrency or p95/p99 production acceptance result. +`/buzz/v1/me/read-state`, plus database pool/statement metrics. No payload, +actor, channel or position values should become metric labels. ## Compatibility and extension rules Within `/buzz/v1`, clients must ignore unknown response object fields. Existing required fields, status variants and their meanings remain stable; additive -fields do not authorize silently changing `attention` or frontier semantics. +fields do not authorize silently changing `unread`, `mentions` or position +semantics. Breaking changes require an explicitly negotiated contract or a new API version. Requests remain strict: send new parameters or intent types only after the relay advertises the corresponding capability. Missing optional data means unsupported @@ -281,8 +255,8 @@ additional capabilities. These typed relational frontiers are signer-private application state, **not self-encrypted**. Database operators can see reading progress; ordinary Nostr queries, search and moderator interfaces do not expose it. No public receipts -are emitted. Storage grows by touched contexts, not observed messages, and has -no fixed context-count ceiling. +are emitted. Storage grows by touched channels and followed threads, not observed +messages, and has no fixed ceiling. Leaving/rejoining does not erase progress; revoked access hides it. Soft-deleted channels are inaccessible, while hard channel deletion cascades their frontiers. diff --git a/migrations/0056_personal_read_state.sql b/migrations/0056_personal_read_state.sql index 3b5ef3a7a6..cc0629af0e 100644 --- a/migrations/0056_personal_read_state.sql +++ b/migrations/0056_personal_read_state.sql @@ -2,7 +2,9 @@ -- A frontier is the relay arrival time (events.received_at) of the message a -- context was read through, never signed event time, which the sender chooses. -- An empty root_id covers only the channel timeline; a root-specific frontier --- covers that thread, without inheritance. +-- covers that thread, without inheritance. A thread row exists for each +-- thread the actor follows or has unfollowed. Ingest moves the author's own +-- frontiers through each message they post. -- threads_through_timestamp is the only cross-context cut: an explicit -- whole-channel read that also covers every thread in that channel. -- started_at is the actor's first read intent and the floor of every @@ -24,6 +26,9 @@ CREATE TABLE personal_read_frontiers ( -- Whole-channel cut covering every thread; channel rows only. threads_through_timestamp TIMESTAMPTZ CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea), + -- Thread rows only: false after unfollow, kept so later replies don't + -- re-follow. Replying or a follow intent sets it again. + following BOOLEAN NOT NULL DEFAULT true CHECK (following OR root_id <> ''::bytea), PRIMARY KEY (community_id, actor, channel_id, root_id), FOREIGN KEY (community_id, actor) REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE, @@ -36,5 +41,16 @@ CREATE TABLE personal_read_frontiers ( -- with nothing new since the reader's position without reading replies. ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ; +-- Arrival time and read scope of each mention, so the sidebar counts a +-- reader's unread mentions in one channel timeline or thread from its +-- position. root_id is the thread a reply belongs to, NULL when the mention +-- shows on the channel timeline (a top-level message or broadcast reply). +-- Existing rows stay NULL: every position is floored at its account's +-- started_at, set by a read intent after this migration, so no earlier +-- mention can be unread. +ALTER TABLE event_mentions ADD COLUMN received_at TIMESTAMPTZ, ADD COLUMN root_id BYTEA; +CREATE INDEX idx_event_mentions_scope_received + ON event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at); + 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 1f3de52f57..1a94faf2da 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -216,6 +216,9 @@ CREATE TABLE personal_read_frontiers ( -- Whole-channel cut covering every thread; channel rows only. threads_through_timestamp TIMESTAMPTZ CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea), + -- Thread rows only: false after unfollow, kept so later replies don't + -- re-follow. Replying or a follow intent sets it again. + following BOOLEAN NOT NULL DEFAULT true CHECK (following OR root_id <> ''::bytea), PRIMARY KEY (community_id, actor, channel_id, root_id), FOREIGN KEY (community_id, actor) REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE, @@ -325,6 +328,10 @@ CREATE TABLE event_mentions ( event_created_at TIMESTAMPTZ NOT NULL, channel_id UUID, event_kind INT, + -- Arrival and read scope for personal read state: root_id is a reply's + -- thread, NULL on the channel timeline. NULL received_at predates it. + received_at TIMESTAMPTZ, + root_id BYTEA, PRIMARY KEY (community_id, pubkey_hex, event_id) ); @@ -332,6 +339,8 @@ CREATE INDEX idx_event_mentions_pubkey_created ON event_mentions (community_id, pubkey_hex, event_created_at DESC); CREATE INDEX idx_event_mentions_pubkey_kind_created ON event_mentions (community_id, pubkey_hex, event_kind, event_created_at DESC); +CREATE INDEX idx_event_mentions_scope_received + ON event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at); -- ── Subscriptions ───────────────────────────────────────────────────────────── -- Conformance: "Mesh, agents, ACP/MCP, and CLI" (persisted subscriptions).