diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 183585e2785..9eec332b44c 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -447,6 +447,16 @@ pub async fn insert_event_in_transaction( ) -> Result<(StoredEvent, bool)> { let result = insert_event_on(tx, event, channel_id).await?; if result.1 { + if let Some(channel) = channel_id { + crate::personal_read::record_message( + tx, + event, + channel, + crate::personal_read::Place::TopLevel, + result.0.received_at, + ) + .await?; + } crate::operator_listener::enqueue_mentions_in_transaction(tx, event).await?; } Ok(result) @@ -1872,6 +1882,22 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( } } + let place = match &thread_meta { + Some(meta) => Some(( + meta.channel_id, + match meta.parent_event_id { + Some(parent) => crate::personal_read::Place::Reply { + root: meta.root_event_id.unwrap_or(parent), + }, + // Workflow messages store a depth-0 metadata row. + None => crate::personal_read::Place::TopLevel, + }, + )), + None => channel_id.map(|channel| (channel, crate::personal_read::Place::TopLevel)), + }; + if let Some((channel, place)) = place { + crate::personal_read::record_message(tx, event, channel, place, received_at).await?; + } crate::operator_listener::enqueue_mentions_in_transaction(tx, 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 a36c2d3c9c4..f8add017bf1 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,9 @@ //! 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::fixture, *}; +use super::{ + postgres_tests::{fixture, insert_reply}, + *, +}; use crate::Db; use buzz_core::CommunityId; use nostr::{EventBuilder, Keys, Kind, Tag}; @@ -45,7 +48,8 @@ async fn post( event } -/// A reply to `root` that mentions the actor, so it counts without membership. +/// A reply to `root` that mentions the actor, so the actor follows the thread +/// from just before the earliest such reply to arrive. #[allow(clippy::too_many_arguments)] async fn reply( db: &Db, @@ -57,12 +61,29 @@ async fn reply( authored: u64, arrived: u64, ) -> nostr::Event { - let event = post(db, pool, community, channel, actor, authored, arrived).await; - sqlx::query("INSERT INTO thread_metadata (community_id,event_id,event_created_at,channel_id,root_event_id,parent_event_id,depth) - VALUES ($1,$2,to_timestamp($3),$4,$5,$5,1)") - .bind(community.as_uuid()).bind(event.id.as_bytes().as_slice()) - .bind(event.created_at.as_secs() as f64) - .bind(channel).bind(root.id.as_bytes().as_slice()).execute(pool).await.unwrap(); + let event = EventBuilder::new(Kind::Custom(9), format!("authored {authored}")) + .tags([Tag::public_key(actor.public_key())]) + .custom_created_at(nostr::Timestamp::from(authored)) + .sign_with_keys(&Keys::generate()) + .unwrap(); + insert_reply(db, community, channel, root, &event).await; + arrive(pool, community, &event, arrived).await; + // Ingest started the follow just before the real arrival; move it to + // just before the rewritten one. + let moved = sqlx::query( + "UPDATE personal_read_frontiers + SET through_timestamp=LEAST(through_timestamp, to_timestamp($4)-interval '1 microsecond') + WHERE community_id=$1 AND actor=$2 AND root_id=$3", + ) + .bind(community.as_uuid()) + .bind(actor.public_key().to_bytes().as_slice()) + .bind(root.id.as_bytes().as_slice()) + .bind(arrived as f64) + .execute(pool) + .await + .unwrap() + .rows_affected(); + assert_eq!(moved, 1, "the mention follows the thread"); event } @@ -233,11 +254,39 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { assert!(row.threads.is_empty()); } -/// The shallow latest probe reads the newest `MAX_CHANNEL_SCAN + 1` events by -/// author time, of any kind. A late arrival with an older author time than -/// that many events is counted unread, so the unread scan must supply the -/// anchor or Mark as read with the sidebar anchor leaves it. -async fn mark_as_read_behind_fillers(fillers: u64) -> (bool, u32) { +/// The latest probe reads the newest `MAX_TIMELINE_SCAN` events by author +/// time, of any kind. A late arrival with an older author time than the +/// uncounted events after it must still be the sidebar anchor, or Mark as read +/// with that anchor leaves it unread. +async fn reactions( + pool: &PgPool, + db: &Db, + community: CommunityId, + channel: Uuid, + count: u64, + now: u64, +) { + for i in 0..count { + // Reactions are not counted, but the probe's LIMIT sees them. + let filler = EventBuilder::new(Kind::Custom(7), "+") + .custom_created_at(nostr::Timestamp::from(now - 500 + i % 400)) + .sign_with_keys(&Keys::generate()) + .unwrap(); + db.insert_event(community, &filler, Some(channel)) + .await + .unwrap(); + } + sqlx::query("UPDATE events SET received_at=to_timestamp($2) WHERE community_id=$1 AND kind=7") + .bind(community.as_uuid()) + .bind((now - 50) as f64) + .execute(pool) + .await + .unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn mark_as_read_clears_a_late_arrival_behind_reactions() { let (db, pool, community, channel, actor, first) = fixture().await; let now = first.created_at.as_secs(); arrive(&pool, community, &first, now - 60).await; @@ -248,17 +297,8 @@ async fn mark_as_read_behind_fillers(fillers: u64) -> (bool, u32) { mark_through(channel, None, &first.id.to_hex()), ) .await; - for i in 0..fillers { - // Reactions are not counted, but the probe's LIMIT sees them. - let filler = EventBuilder::new(Kind::Custom(7), "+") - .custom_created_at(nostr::Timestamp::from(now - 500 + i)) - .sign_with_keys(&Keys::generate()) - .unwrap(); - db.insert_event(community, &filler, Some(channel)) - .await - .unwrap(); - arrive(&pool, community, &filler, now - 50).await; - } + // More than main's old 256-event probe. + reactions(&pool, &db, community, channel, 300, now).await; post(&db, &pool, community, channel, &actor, now - 600, now - 10).await; assert_eq!(counts(&db, community, &actor).await, (true, 1)); @@ -267,29 +307,14 @@ async fn mark_as_read_behind_fillers(fillers: u64) -> (bool, u32) { .latest_id .expect("a channel with messages has a latest message"); apply(&db, community, &actor, mark_through(channel, None, &anchor)).await; - counts(&db, community, &actor).await -} - -/// Control: with one slot to spare the late arrival is inside the probe. -#[tokio::test] -#[ignore = "requires Postgres"] -async fn mark_as_read_clears_a_late_arrival_inside_the_latest_probe() { - let fillers = MAX_CHANNEL_SCAN as u64 - 1; - assert_eq!(mark_as_read_behind_fillers(fillers).await, (false, 0)); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn mark_as_read_clears_a_late_arrival_outside_the_latest_probe() { - let fillers = MAX_CHANNEL_SCAN as u64; - assert_eq!(mark_as_read_behind_fillers(fillers).await, (false, 0)); + assert_eq!(counts(&db, community, &actor).await, (false, 0)); } -/// When the newest 257 events are all uncounted, the only message is still in -/// the unread scan: the summary reports it as latest and unread. +/// Past the budget the probe finds no anchor, and nothing counts: a badge +/// with no latest to mark could never be cleared. #[tokio::test] #[ignore = "requires Postgres"] -async fn latest_message_behind_a_full_probe_of_reactions_is_found() { +async fn a_message_behind_a_full_budget_of_reactions_is_not_a_badge_without_an_anchor() { 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") @@ -300,21 +325,20 @@ async fn latest_message_behind_a_full_probe_of_reactions_is_found() { .unwrap() .rows_affected(); assert_eq!(demoted, 1); - let only = post(&db, &pool, community, channel, &actor, now - 600, now - 10).await; - for i in 0..MAX_CHANNEL_SCAN as u64 { - let filler = EventBuilder::new(Kind::Custom(7), "+") - .custom_created_at(nostr::Timestamp::from(now - 500 + i)) - .sign_with_keys(&Keys::generate()) - .unwrap(); - db.insert_event(community, &filler, Some(channel)) - .await - .unwrap(); - arrive(&pool, community, &filler, now - 50).await; - } + post(&db, &pool, community, channel, &actor, now - 600, now - 10).await; + reactions( + &pool, + &db, + community, + channel, + MAX_TIMELINE_SCAN as u64, + now, + ) + .await; let row = sidebar(&db, community, &actor).await; - assert_eq!(row.latest_id, Some(only.id.to_hex())); - assert_eq!((row.unread, row.mentions), (true, 1)); + assert_eq!(row.latest_id, None); + assert_eq!((row.unread, row.mentions), (false, 0)); } /// Store `event` as having arrived at exactly `seconds` plus `micros`. Built diff --git a/crates/buzz-db/src/store/personal_read/follows.rs b/crates/buzz-db/src/store/personal_read/follows.rs new file mode 100644 index 00000000000..680ccfb9dcc --- /dev/null +++ b/crates/buzz-db/src/store/personal_read/follows.rs @@ -0,0 +1,113 @@ +//! Read state that ingest writes. A thread frontier row exists only for a +//! thread its actor follows; only this module creates one. +use super::model::ELIGIBLE_KINDS; +use crate::{AdmittedTx, Result}; +use chrono::{DateTime, Utc}; +use nostr::Event; +use uuid::Uuid; + +/// Where a newly stored channel message sits. +#[derive(Clone, Copy)] +pub(crate) enum Place<'a> { + /// A top-level timeline message. + TopLevel, + /// A reply in the thread rooted at `root`. + Reply { root: &'a [u8] }, +} + +/// Record what a newly stored eligible message changes, in its transaction: +/// - its author has read through it: the timeline for a top-level message, +/// otherwise the thread, which the author then follows; +/// - each channel member it mentions follows its thread (for a top-level +/// message, the thread rooted at it); +/// - a reply makes the root's author, and in a DM every member, follow too. +/// +/// Everyone but the author starts just before the message, so it is unread +/// to them. A follower keeps an existing position. +pub(crate) async fn record_message( + tx: &mut AdmittedTx, + event: &Event, + channel: Uuid, + place: Place<'_>, + received_at: DateTime, +) -> Result<()> { + if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) { + return Ok(()); + } + let community = tx.community(); + let id = event.id.as_bytes().as_slice(); + let mentioned: Vec> = event + .tags + .public_keys() + .map(|key| key.to_bytes().to_vec()) + .collect(); + let (read, thread, reply) = match place { + Place::TopLevel => (&[][..], id, false), + Place::Reply { root } => (root, root, true), + }; + // Accounts, then frontiers, each in key order, so concurrent messages that + // involve the same people cannot deadlock: the frontier insert joins the + // finished account insert. Foreign-key checks run at the end of the + // statement, after the account rows exist. + sqlx::query( + "WITH involved (actor, root_id, through, message) AS ( + SELECT $4::bytea, $5::bytea, $6::timestamptz, $3::bytea + UNION ALL + SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL FROM channel_members cm + WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL + AND cm.pubkey=ANY($8) AND cm.pubkey<>$4 + UNION ALL + SELECT e.pubkey, $7, $6-interval '1 microsecond', NULL FROM events e + WHERE $9 AND e.community_id=$1 AND e.channel_id=$2 AND e.id=$7 AND e.pubkey<>$4 + UNION ALL + SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL + FROM channel_members cm JOIN channels c + ON c.community_id=cm.community_id AND c.id=cm.channel_id + WHERE $9 AND cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL + AND c.channel_type='dm' AND cm.pubkey<>$4 + ), rows AS ( + -- An existing follower keeps their position, so is not touched. + -- Existence probes are LIMIT 1 laterals: always one index lookup. + SELECT DISTINCT ON (i.actor, i.root_id) i.* FROM involved i + LEFT JOIN LATERAL ( + SELECT true AS found FROM personal_read_frontiers f WHERE f.community_id=$1 + AND f.actor=i.actor AND f.channel_id=$2 AND f.root_id=i.root_id LIMIT 1 + ) f ON true + WHERE i.message IS NOT NULL OR f.found IS NULL + ORDER BY i.actor, i.root_id + ), accounts AS ( + INSERT INTO personal_read_accounts (community_id, actor) + SELECT $1, r.actor FROM (SELECT DISTINCT actor FROM rows) r + LEFT JOIN LATERAL ( + SELECT true AS found FROM personal_read_accounts a + WHERE a.community_id=$1 AND a.actor=r.actor LIMIT 1 + ) a ON true + WHERE a.found IS NULL ORDER BY r.actor + ON CONFLICT DO NOTHING RETURNING 1 + ) + INSERT INTO personal_read_frontiers + (community_id, actor, channel_id, root_id, through_timestamp, through_message_id) + SELECT $1, actor, $2, root_id, through, message + FROM rows CROSS JOIN (SELECT count(*) FROM accounts) written + ORDER BY actor, root_id + ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE + SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, + excluded.through_timestamp), + through_message_id=CASE + WHEN excluded.through_timestamp > personal_read_frontiers.through_timestamp + THEN excluded.through_message_id ELSE personal_read_frontiers.through_message_id END + WHERE excluded.through_message_id IS NOT NULL", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(id) + .bind(event.pubkey.to_bytes().as_slice()) + .bind(read) + .bind(received_at) + .bind(thread) + .bind(&mentioned) + .bind(reply) + .execute(tx.conn()) + .await?; + Ok(()) +} diff --git a/crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs new file mode 100644 index 00000000000..33a2e75929e --- /dev/null +++ b/crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs @@ -0,0 +1,373 @@ +//! Who follows a thread, as recorded by ingest, and what following shows. +use super::{ + postgres_tests::{fixture, insert_reply, start_before_everything}, + *, +}; +use crate::Db; +use buzz_core::{channel::MemberRole, CommunityId}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use sqlx::PgPool; +use uuid::Uuid; + +fn message(author: &Keys, tags: Vec) -> nostr::Event { + EventBuilder::new(Kind::Custom(9), Uuid::new_v4().to_string()) + .tags(tags) + .sign_with_keys(author) + .unwrap() +} + +async fn post( + db: &Db, + community: CommunityId, + channel: Uuid, + author: &Keys, + tags: Vec, +) -> nostr::Event { + let event = message(author, tags); + db.insert_event(community, &event, Some(channel)) + .await + .unwrap(); + event +} + +async fn reply( + db: &Db, + community: CommunityId, + channel: Uuid, + root: &nostr::Event, + author: &Keys, + tags: Vec, +) -> nostr::Event { + let event = message(author, tags); + insert_reply(db, community, channel, root, &event).await; + event +} + +fn mention(who: &Keys) -> Vec { + vec![Tag::public_key(who.public_key())] +} + +async fn member(db: &Db, pool: &PgPool, community: CommunityId, channel: Uuid) -> Keys { + let keys = Keys::generate(); + db.add_member( + community, + channel, + &keys.public_key().to_bytes(), + MemberRole::Member, + None, + ) + .await + .unwrap(); + start_before_everything(pool, community, &keys.public_key()).await; + keys +} + +async fn row(db: &Db, community: CommunityId, channel: Uuid, actor: &Keys) -> ChannelReadSummary { + db.personal_read_sidebar_channels( + community, + &actor.public_key(), + DEFAULT_RETENTION_SECONDS, + &[channel], + ) + .await + .unwrap() + .channels + .remove(0) +} + +/// The actor's frontier message for a thread, if the actor follows it. A +/// follower who has read nothing yet has no message. +async fn follow( + pool: &PgPool, + community: CommunityId, + actor: &Keys, + root: &nostr::Event, +) -> Option> { + sqlx::query_scalar( + "SELECT encode(through_message_id,'hex') FROM personal_read_frontiers + WHERE community_id=$1 AND actor=$2 AND root_id=$3", + ) + .bind(community.as_uuid()) + .bind(actor.public_key().to_bytes().as_slice()) + .bind(root.id.as_bytes().as_slice()) + .fetch_optional(pool) + .await + .unwrap() +} + +/// (root, unread replies) for each listed thread. +fn threads(row: &ChannelReadSummary) -> Vec<(String, u32)> { + row.threads + .iter() + .map(|t| (t.root_id.clone(), t.mentions)) + .collect() +} + +fn mark_thread(channel: Uuid, root: &nostr::Event, message: &str) -> ReadIntent { + ReadIntent::MarkThrough { + target: ReadTarget { + channel_id: channel, + root_id: Some(root.id.to_hex()), + }, + message_id: message.to_owned(), + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn replying_follows_the_thread_and_reads_through_the_reply() { + let (db, pool, community, channel, actor, root) = fixture().await; + let own = reply(&db, community, channel, &root, &actor, vec![]).await; + assert_eq!( + follow(&pool, community, &actor, &root).await, + Some(Some(own.id.to_hex())) + ); + assert!(row(&db, community, channel, &actor) + .await + .threads + .is_empty()); + + let other = reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + let summary = row(&db, community, channel, &actor).await; + assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]); + assert_eq!(summary.threads[0].latest_id, other.id.to_hex()); + assert_eq!(summary.threads[0].read_through_id, Some(own.id.to_hex())); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_mention_follows_a_member_from_just_before_it() { + let (db, pool, community, channel, _, root) = fixture().await; + let bee = member(&db, &pool, community, channel).await; + let outsider = Keys::generate(); + let earlier = reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + let mut tags = mention(&bee); + tags.extend(mention(&outsider)); + let mentioning = reply(&db, community, channel, &root, &Keys::generate(), tags).await; + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + + assert_eq!(follow(&pool, community, &bee, &root).await, Some(None)); + assert_eq!(follow(&pool, community, &outsider, &root).await, None); + // The mention and what came after it; not the reply before it. + let summary = row(&db, community, channel, &bee).await; + assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]); + assert_ne!(summary.threads[0].latest_id, earlier.id.to_hex()); + assert_ne!(summary.threads[0].latest_id, mentioning.id.to_hex()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_later_mention_keeps_the_followers_position() { + let (db, pool, community, channel, actor, root) = fixture().await; + let own = reply(&db, community, channel, &root, &actor, vec![]).await; + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + reply( + &db, + community, + channel, + &root, + &Keys::generate(), + mention(&actor), + ) + .await; + assert_eq!( + follow(&pool, community, &actor, &root).await, + Some(Some(own.id.to_hex())) + ); + let summary = row(&db, community, channel, &actor).await; + assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn the_root_author_follows_on_the_first_reply() { + let (db, pool, community, channel, actor, _) = fixture().await; + let root = post(&db, community, channel, &actor, vec![]).await; + assert_eq!(follow(&pool, community, &actor, &root).await, None); + + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + assert_eq!(follow(&pool, community, &actor, &root).await, Some(None)); + let summary = row(&db, community, channel, &actor).await; + assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_top_level_mention_follows_the_thread_rooted_at_it() { + let (db, pool, community, channel, actor, _) = fixture().await; + let root = post(&db, community, channel, &Keys::generate(), mention(&actor)).await; + assert_eq!(follow(&pool, community, &actor, &root).await, Some(None)); + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + + let summary = row(&db, community, channel, &actor).await; + // The fixture message and the mention are unread; one is a mention. + assert_eq!((summary.unread, summary.mentions), (true, 1)); + assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dm_members_follow_every_thread() { + let (db, pool, community, _, actor, _) = fixture().await; + let (bee, wasp) = (Keys::generate(), Keys::generate()); + let keys = [&actor, &bee, &wasp].map(|k| k.public_key().to_bytes()); + let participants: Vec<&[u8]> = keys.iter().map(|k| k.as_slice()).collect(); + let dm = db + .create_dm(community, &participants, &keys[1]) + .await + .unwrap() + .id; + let root = post(&db, community, dm, &bee, vec![]).await; + reply(&db, community, dm, &root, &wasp, vec![]).await; + + assert_eq!(follow(&pool, community, &actor, &root).await, Some(None)); + let summary = row(&db, community, dm, &actor).await; + // Every DM message counts as a mention. + assert_eq!((summary.unread, summary.mentions), (true, 1)); + assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_unfollowed_thread_never_counts_and_marking_it_follows_nothing() { + let (db, pool, community, channel, actor, root) = fixture().await; + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + let broadcast = vec![Tag::parse(["broadcast", "1"]).unwrap()]; + let last = reply(&db, community, channel, &root, &Keys::generate(), broadcast).await; + let summary = row(&db, community, channel, &actor).await; + assert!(summary.threads.is_empty()); + // Replies are not on the timeline; only the fixture message is unread. + assert_eq!((summary.unread, summary.mentions), (true, 0)); + + let outcome = db + .apply_personal_read_intent( + community, + &actor.public_key(), + &mark_thread(channel, &root, &last.id.to_hex()), + ) + .await + .unwrap(); + assert_eq!(outcome, IntentOutcome::Applied); + assert_eq!(follow(&pool, community, &actor, &root).await, None); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn posting_reads_the_timeline_through_the_post() { + let (db, _pool, community, channel, actor, _) = fixture().await; + assert!(row(&db, community, channel, &actor).await.unread); + let own = post(&db, community, channel, &actor, vec![]).await; + let summary = row(&db, community, channel, &actor).await; + assert!(!summary.unread); + assert_eq!(summary.read_through_id, Some(own.id.to_hex())); + assert_eq!(summary.latest_id, Some(own.id.to_hex())); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn thread_counts_stop_at_the_cap() { + let (db, _pool, community, channel, actor, root) = fixture().await; + reply(&db, community, channel, &root, &actor, vec![]).await; + let other = Keys::generate(); + for _ in 0..=MAX_UNREAD_COUNT { + reply(&db, community, channel, &root, &other, vec![]).await; + } + let summary = row(&db, community, channel, &actor).await; + assert_eq!( + threads(&summary), + [(root.id.to_hex(), MAX_UNREAD_COUNT as u32)] + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn ingest_does_not_start_an_account() { + let (db, pool, community, channel, _, root) = fixture().await; + let bee = Keys::generate(); + db.add_member( + community, + channel, + &bee.public_key().to_bytes(), + MemberRole::Member, + None, + ) + .await + .unwrap(); + reply( + &db, + community, + channel, + &root, + &Keys::generate(), + mention(&bee), + ) + .await; + reply(&db, community, channel, &root, &bee, vec![]).await; + + let started: Option> = sqlx::query_scalar( + "SELECT started_at FROM personal_read_accounts WHERE community_id=$1 AND actor=$2", + ) + .bind(community.as_uuid()) + .bind(bee.public_key().to_bytes().as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(started, None); + // Before its first intent, nothing counts. + let summary = row(&db, community, channel, &bee).await; + assert_eq!((summary.unread, summary.mentions), (false, 0)); + assert!(summary.threads.is_empty()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_nested_reply_follows_the_root_thread() { + let (db, pool, community, channel, actor, _) = fixture().await; + let root = post(&db, community, channel, &actor, vec![]).await; + let parent = reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + let bee = Keys::generate(); + let at = |e: &nostr::Event| { + chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap() + }; + let nested = message(&bee, vec![]); + db.insert_event_with_thread_metadata( + community, + &nested, + Some(channel), + Some(crate::event::ThreadMetadataParams { + event_id: nested.id.as_bytes(), + event_created_at: at(&nested), + channel_id: channel, + parent_event_id: Some(parent.id.as_bytes()), + parent_event_created_at: Some(at(&parent)), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(at(&root)), + depth: 2, + broadcast: false, + }), + ) + .await + .unwrap(); + + assert_eq!( + follow(&pool, community, &bee, &root).await, + Some(Some(nested.id.to_hex())) + ); + assert_eq!(follow(&pool, community, &bee, &parent).await, None); + let summary = row(&db, community, channel, &actor).await; + assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]); + assert_eq!(summary.threads[0].latest_id, nested.id.to_hex()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_message_stored_without_thread_metadata_reads_the_timeline() { + let (db, _pool, community, channel, actor, _) = fixture().await; + let own = message(&actor, vec![]); + db.insert_event_with_thread_metadata(community, &own, Some(channel), None) + .await + .unwrap(); + let summary = row(&db, community, channel, &actor).await; + assert!(!summary.unread); + assert_eq!(summary.read_through_id, Some(own.id.to_hex())); +} diff --git a/crates/buzz-db/src/store/personal_read/mod.rs b/crates/buzz-db/src/store/personal_read/mod.rs index bd01af0cdeb..130b55d81c4 100644 --- a/crates/buzz-db/src/store/personal_read/mod.rs +++ b/crates/buzz-db/src/store/personal_read/mod.rs @@ -1,21 +1,23 @@ //! 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; the unread horizon alone uses author time. Only fixed context -//! intents advance frontiers, never a query scan cap. +//! through; the unread horizon alone uses author time. Read intents and the +//! author's own posts advance frontiers. A thread frontier exists only for a +//! followed thread, and the sidebar counts forward from each frontier. +mod follows; mod model; -mod participation; mod projection; mod writes; +pub(crate) use follows::{record_message, Place}; pub use model::*; #[cfg(test)] mod postgres_tests; #[cfg(test)] -mod participation_postgres_tests; +mod follows_postgres_tests; #[cfg(test)] mod projection_postgres_tests; diff --git a/crates/buzz-db/src/store/personal_read/model.rs b/crates/buzz-db/src/store/personal_read/model.rs index 448d2491cb3..1b276918f6c 100644 --- a/crates/buzz-db/src/store/personal_read/model.rs +++ b/crates/buzz-db/src/store/personal_read/model.rs @@ -57,10 +57,17 @@ pub const MAX_CHANNELS: usize = 20; /// Maximum thread rows per channel row. A POST can return rows for /// `MAX_INTENTS` channels, and they must fit the 1 MiB response limit. pub const MAX_THREAD_SUMMARIES: usize = 25; -/// Latest-probe event budget per channel, before eligibility filtering. -pub const MAX_CHANNEL_SCAN: usize = 256; -/// Unread-window work budget per channel, before eligibility/ancestry joins. -pub const MAX_UNREAD_SCAN: usize = 4096; +/// Events each walk examines before eligibility: newest-first for the +/// newest timeline message, then for the last arrival among those authored up +/// to twice the skew before it (in a thread too), and forward from the read +/// position for DM counts. What lies past the budget goes unseen. +pub const MAX_TIMELINE_SCAN: usize = 4096; +/// Counts stop here: a count of 100 means at least 100. +pub const MAX_UNREAD_COUNT: usize = 100; +/// The relay accepts author times up to this far from arrival either way. +/// Read positions are arrival times and the indexes are by author time, so +/// every forward walk starts this much before the position. +pub const MAX_ARRIVAL_SKEW_SECONDS: u32 = 15 * 60; /// Conversation kinds eligible for ordinary unread state (not edits/reactions). pub const ELIGIBLE_KINDS: [i32; 4] = [9, 40002, 45001, 45003]; @@ -74,32 +81,31 @@ pub struct ChannelReadSummary { /// only on their thread row. pub unread: bool, /// Unread top-level messages directed at the actor: every one in a DM, - /// otherwise those that tag the actor with `p` or carry `broadcast=1`. + /// otherwise those that tag the actor with `p`. pub mentions: u32, - /// The anchor of the timeline's frontier: the marked message that arrived - /// last. None until the timeline is marked. + /// The anchor of the timeline's frontier: the marked or own top-level + /// message that arrived last. None until either exists. pub read_through_id: Option, /// Last eligible top-level message to arrive, whatever its author or read /// progress: marking the timeline through it reads the timeline. pub latest_id: Option, - /// Threads with unread replies, newest unread reply first. + /// Followed threads with unread replies, newest unread reply first. pub threads: Vec, } /// One thread's read state within a channel row. No conversation bytes. -#[derive(Debug, Serialize)] +#[derive(Debug, Deserialize, Serialize)] pub struct ThreadReadSummary { /// Canonical thread-root event ID. pub root_id: String, /// Whether an unread reply that counts was found. pub unread: bool, - /// Unread replies that count: directed at the actor, or in one of the - /// actor's conversations. + /// Unread replies by others in this followed thread. pub mentions: u32, - /// The anchor of the thread's frontier: the marked message that arrived - /// last. None until the thread is marked. + /// The anchor of the thread's frontier: the marked or own reply that + /// arrived last. None for a follower who has read nothing yet. pub read_through_id: Option, - /// Last counted reply to arrive: marking through it reads the thread. + /// Last reply to arrive: marking through it reads the thread. pub latest_id: String, } diff --git a/crates/buzz-db/src/store/personal_read/participation.rs b/crates/buzz-db/src/store/personal_read/participation.rs deleted file mode 100644 index 682bca30489..00000000000 --- a/crates/buzz-db/src/store/personal_read/participation.rs +++ /dev/null @@ -1,127 +0,0 @@ -//! Conversation membership from the existing thread store, bounded by time. -use super::model::ELIGIBLE_KINDS; -use crate::Result; -use buzz_core::CommunityId; -use sqlx::{Acquire, PgConnection, Row}; -use std::collections::HashMap; -use uuid::Uuid; - -// Bound multiplicative work independently of the unread evidence window. -const MAX_PARENTS: usize = 1024; - -/// Whether each `(channel, parent)` is one of the actor's conversations: the -/// actor wrote the parent or has a reply to it. Only live eligible messages -/// qualify. An absent key is undecided: past the target cap, or the statement -/// deadline expired. Membership is independent of unread retention and read -/// frontiers. -pub(super) async fn resolve( - conn: &mut PgConnection, - community: CommunityId, - actor: &[u8], - targets: &[(Uuid, Vec)], -) -> Result), bool>> { - if targets.is_empty() { - return Ok(HashMap::new()); - } - let targets = select_targets(targets); - let channels: Vec<_> = targets.iter().map(|(channel, _)| *channel).collect(); - let parents: Vec<_> = targets.iter().map(|(_, parent)| parent.clone()).collect(); - // Optional inference must not abort authoritative unread/frontier reads. - // A nested transaction is a savepoint; rollback also restores the caller's - // statement timeout. No state is written in this read-only inference. - let mut budget = conn.begin().await?; - sqlx::query("SET LOCAL statement_timeout = '500ms'") - .execute(&mut *budget) - .await?; - sqlx::query("SET LOCAL jit = off") - .execute(&mut *budget) - .await?; - // Exact: a row cap would lose a real member behind a busy parent's other - // replies. LATERAL ... LIMIT 1 keeps the work on the replies under the - // requested parents; a plain EXISTS may be hashed over the whole tenant. - let result = sqlx::query( - "SELECT t.channel_id,t.parent_id,(own.hit OR replied.hit) IS TRUE AS member - FROM unnest($3::uuid[],$4::bytea[]) t(channel_id,parent_id) - LEFT JOIN LATERAL ( - SELECT e.pubkey=$2 AND e.deleted_at IS NULL AND e.kind=ANY($5) AS hit - FROM events e WHERE e.community_id=$1 AND e.channel_id=t.channel_id AND e.id=t.parent_id - ORDER BY e.created_at DESC LIMIT 1 - ) own ON true - LEFT JOIN LATERAL ( - SELECT true AS hit FROM thread_metadata tm JOIN events e ON e.community_id=$1 - AND e.channel_id=t.channel_id AND e.created_at=tm.event_created_at AND e.id=tm.event_id - WHERE tm.community_id=$1 AND tm.channel_id=t.channel_id AND tm.parent_event_id=t.parent_id - AND e.pubkey=$2 AND e.deleted_at IS NULL AND e.kind=ANY($5) - AND own.hit IS NOT TRUE - LIMIT 1 - ) replied ON true", - ) - .bind(community.as_uuid()) - .bind(actor) - .bind(channels) - .bind(parents) - .bind(ELIGIBLE_KINDS.as_slice()) - .fetch_all(&mut *budget) - .await; - budget.rollback().await?; - let rows = match result { - Ok(rows) => rows, - Err(sqlx::Error::Database(error)) - if matches!(error.code().as_deref(), Some("57014" | "55P03")) => - { - return Ok(HashMap::new()); - } - Err(error) => return Err(error.into()), - }; - rows.into_iter() - .map(|row| { - Ok(( - (row.try_get("channel_id")?, row.try_get("parent_id")?), - row.try_get("member")?, - )) - }) - .collect() -} - -// Keep selection independent of the optional SQL deadline: a timeout must not -// hide a regression in the cardinality bound. -fn select_targets(targets: &[(Uuid, Vec)]) -> Vec<(Uuid, Vec)> { - let mut targets = targets.to_vec(); - targets.sort_unstable(); - targets.dedup(); - targets.truncate(MAX_PARENTS); - targets -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn target_selection_caps_unique_parents_independently_of_sql_timeout() { - // Literal contract boundaries deliberately do not derive from MAX_PARENTS. - for count in [0_u32, 1, 1023, 1024, 1025] { - let unique: Vec<_> = (0..count) - .map(|i| (Uuid::nil(), i.to_be_bytes().to_vec())) - .collect(); - let input: Vec<_> = unique - .iter() - .rev() - .chain(unique.iter().rev()) - .cloned() - .collect(); - let selected = select_targets(&input); - assert_eq!(selected, unique[..unique.len().min(1024)], "count {count}"); - } - } - - #[test] - fn target_selection_keeps_channel_identity() { - let first = (Uuid::from_u128(1), vec![7; 32]); - let second = (Uuid::from_u128(2), vec![7; 32]); - assert_eq!( - select_targets(&[second.clone(), first.clone(), second.clone()]), - vec![first, second] - ); - } -} diff --git a/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs deleted file mode 100644 index 5a3d3a0acad..00000000000 --- a/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs +++ /dev/null @@ -1,537 +0,0 @@ -//! Which replies count: the direct-parent conversation rule, its deletion and -//! channel edges, and the bounds on the membership lookup. -use super::*; -use crate::{ - channel::{ChannelType, ChannelVisibility}, - Db, -}; -use buzz_core::CommunityId; -use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; -use serde_json::{json, Value}; -use sqlx::PgPool; -use uuid::Uuid; - -const DAY: u64 = 86_400; - -/// One community with the reading actor and a peer. -struct World { - db: Db, - pool: PgPool, - community: CommunityId, - actor: Keys, - peer: Keys, - now: u64, -} - -impl World { - async fn new() -> Self { - let pool = PgPool::connect(&crate::test_support::database_url()) - .await - .unwrap(); - let db = Db::from_pool(pool.clone()); - let community = db - .ensure_configured_community(&format!("conversation-{}.local", Uuid::new_v4())) - .await - .unwrap() - .id; - let actor = Keys::generate(); - super::postgres_tests::start_before_everything(&pool, community, &actor.public_key()).await; - Self { - db, - pool, - community, - actor, - peer: Keys::generate(), - now: Timestamp::now().as_secs(), - } - } - - /// A channel the actor has joined. - async fn channel(&self) -> Uuid { - self.db - .create_channel( - self.community, - &Uuid::new_v4().to_string(), - ChannelType::Stream, - ChannelVisibility::Open, - None, - &self.actor.public_key().to_bytes(), - None, - ) - .await - .unwrap() - .id - } - - /// A top-level message. - async fn post(&self, channel: Uuid, author: &Keys, at: u64, tags: Vec) -> nostr::Event { - let event = EventBuilder::new(Kind::Custom(9), Uuid::new_v4().to_string()) - .tags(tags) - .custom_created_at(Timestamp::from(at)) - .sign_with_keys(author) - .unwrap(); - self.db - .insert_event(self.community, &event, Some(channel)) - .await - .unwrap(); - event - } - - /// A canonical reply to `parent` in `root`'s thread. `parent` need not be - /// stored in `channel`: thread metadata records what the reply claims. - async fn reply( - &self, - channel: Uuid, - author: &Keys, - root: &nostr::Event, - parent: Option<&nostr::Event>, - at: u64, - tags: Vec, - ) -> nostr::Event { - let event = self.post(channel, author, at, tags).await; - sqlx::query("INSERT INTO thread_metadata (community_id,event_id,event_created_at,channel_id,root_event_id,parent_event_id,depth) - VALUES ($1,$2,to_timestamp($3),$4,$5,$6,1)") - .bind(self.community.as_uuid()).bind(event.id.as_bytes().as_slice()).bind(at as f64) - .bind(channel).bind(root.id.as_bytes().as_slice()) - .bind(parent.map(|parent| parent.id.as_bytes().as_slice())) - .execute(&self.pool).await.unwrap(); - event - } - - /// The storage write that author and staff deletions share. - async fn delete(&self, event: &nostr::Event) { - assert!(self - .db - .soft_delete_event_and_update_thread(self.community, event.id.as_bytes(), None, None) - .await - .unwrap()); - } - - async fn row(&self, channel: Uuid) -> ChannelReadSummary { - self.db - .personal_read_sidebar_channels( - self.community, - &self.actor.public_key(), - DEFAULT_RETENTION_SECONDS, - &[channel], - ) - .await - .unwrap() - .channels - .remove(0) - } - - /// The row's counts and thread rows: what the sidebar says counts. - async fn counts(&self, channel: Uuid) -> Value { - let row = self.row(channel).await; - json!({"unread":row.unread,"mentions":row.mentions,"threads":wire(&row.threads)}) - } - - fn mention(&self) -> Tag { - Tag::parse(["p", &self.actor.public_key().to_hex()]).unwrap() - } -} - -fn wire(value: &T) -> Value { - serde_json::to_value(value).unwrap() -} - -/// Expected `counts`: whether a top-level message is unread, how many of those -/// are directed, and the listed thread rows. -fn counts(unread: bool, mentions: u32, threads: Vec) -> Value { - json!({"unread":unread,"mentions":mentions,"threads":threads}) -} - -fn broadcast() -> Tag { - Tag::parse(["broadcast", "1"]).unwrap() -} - -/// A listed, never-marked thread with `mentions` counted unread replies. -fn item(root: &nostr::Event, mentions: u32, latest: &nostr::Event) -> Value { - json!({"root_id":root.id.to_hex(),"unread":true,"mentions":mentions, - "read_through_id":null,"latest_id":latest.id.to_hex()}) -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn a_reply_counts_only_in_a_conversation_the_actor_wrote_or_replied_to() { - let w = World::new().await; - // Every witness predates the unread horizon: membership does not expire. - let old = w.now - 40 * DAY; - - // The actor wrote the root. A peer answers it; another peer reply continues - // under that answer, where the actor has written nothing. - let c = w.channel().await; - let root = w.post(c, &w.actor, old, vec![]).await; - let answer = w - .reply(c, &w.peer, &root, Some(&root), w.now + 1, vec![]) - .await; - let nested = w - .reply(c, &w.peer, &root, Some(&answer), w.now + 2, vec![]) - .await; - // Only `answer` counts. The newer reply that does not count is not the - // thread's latest. - assert_eq!( - w.counts(c).await, - counts(false, 0, vec![item(&root, 1, &answer)]) - ); - - // Joining the nested conversation makes its earlier reply count. The - // actor's own reply adds nothing. - w.reply(c, &w.actor, &root, Some(&answer), w.now + 3, vec![]) - .await; - assert_eq!( - w.counts(c).await, - counts(false, 0, vec![item(&root, 2, &nested)]) - ); - - // A peer's thread. The actor replied to one of two parents, long ago. - let c = w.channel().await; - let root = w.post(c, &w.peer, old, vec![]).await; - let joined = w.reply(c, &w.peer, &root, Some(&root), old, vec![]).await; - let other = w.reply(c, &w.peer, &root, Some(&root), old, vec![]).await; - w.reply(c, &w.actor, &root, Some(&joined), old, vec![]) - .await; - let sibling = w - .reply(c, &w.peer, &root, Some(&joined), w.now + 1, vec![]) - .await; - w.reply(c, &w.peer, &root, Some(&other), w.now + 2, vec![]) - .await; - assert_eq!( - w.counts(c).await, - counts(false, 0, vec![item(&root, 1, &sibling)]) - ); - - // A conversation the actor never joined: one unread top-level message, and - // a reply that is not unread even in its own thread. - let c = w.channel().await; - let root = w.post(c, &w.peer, w.now, vec![]).await; - w.reply(c, &w.peer, &root, Some(&root), w.now + 1, vec![]) - .await; - assert_eq!(w.counts(c).await, counts(true, 0, vec![])); - - // A reply whose parent was never recorded is undecided: left out. - w.reply(c, &w.peer, &root, None, w.now + 2, vec![]).await; - assert_eq!(w.counts(c).await, counts(true, 0, vec![])); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn a_directed_reply_counts_outside_the_actors_conversations() { - let w = World::new().await; - let c = w.channel().await; - let root = w.post(c, &w.peer, w.now, vec![]).await; - let parent = w.reply(c, &w.peer, &root, Some(&root), w.now, vec![]).await; - w.reply(c, &w.actor, &root, Some(&root), w.now + 1, vec![]) - .await; - for (at, tags) in [(2, vec![w.mention()]), (3, vec![broadcast()])] { - w.reply(c, &w.peer, &root, Some(&parent), w.now + at, tags) - .await; - } - let both = w - .reply( - c, - &w.peer, - &root, - Some(&parent), - w.now + 4, - vec![broadcast(), w.mention()], - ) - .await; - // The root is an undirected top-level message. The thread counts the - // actor's sibling `parent` and the three directed replies, not the - // actor's own. - let expected = counts(true, 0, vec![item(&root, 4, &both)]); - assert_eq!(w.counts(c).await, expected); - - // Joining the directed replies' conversation changes no count. - w.reply(c, &w.actor, &root, Some(&parent), w.now + 5, vec![]) - .await; - assert_eq!(w.counts(c).await, expected); - - // In a DM every peer message is direct, tagged or not, joined or not. - let dm = w.channel().await; - sqlx::query("UPDATE channels SET channel_type='dm' WHERE community_id=$1 AND id=$2") - .bind(w.community.as_uuid()) - .bind(dm) - .execute(&w.pool) - .await - .unwrap(); - let root = w.post(dm, &w.peer, w.now, vec![w.mention()]).await; - let parent = w - .reply(dm, &w.peer, &root, Some(&root), w.now, vec![]) - .await; - let reply = w - .reply(dm, &w.peer, &root, Some(&parent), w.now + 1, vec![]) - .await; - assert_eq!( - w.counts(dm).await, - counts(true, 1, vec![item(&root, 2, &reply)]) - ); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn eligibility_frontier_and_membership_each_leave_replies_out() { - let w = World::new().await; - let c = w.channel().await; - let root = w.post(c, &w.peer, w.now, vec![]).await; - let parent = w.reply(c, &w.peer, &root, Some(&root), w.now, vec![]).await; - let reply = |author, at, tags| w.reply(c, author, &root, Some(&parent), w.now + at, tags); - // Peer replies to a parent the actor neither wrote nor replied to. - reply(&w.peer, 1, vec![]).await; - let deleted = reply(&w.peer, 2, vec![w.mention()]).await; - w.delete(&deleted).await; - w.reply(c, &w.actor, &root, Some(&root), w.now + 3, vec![]) - .await; - let anchor = reply(&w.peer, 4, vec![w.mention()]).await; - // Undirected, outside the actor's conversations: never counts. - reply(&w.peer, 5, vec![]).await; - // `parent` (a reply to the root, which the actor answered) and `anchor` - // count: not the deleted, own or undirected replies. - assert_eq!( - w.counts(c).await, - counts(true, 0, vec![item(&root, 2, &anchor)]) - ); - - assert_eq!( - w.db.apply_personal_read_intent( - w.community, - &w.actor.public_key(), - &ReadIntent::MarkThrough { - target: ReadTarget { - channel_id: c, - root_id: Some(root.id.to_hex()), - }, - message_id: anchor.id.to_hex(), - }, - ) - .await - .unwrap(), - IntentOutcome::Applied - ); - assert_eq!(w.counts(c).await, counts(true, 0, vec![])); - // A later directed reply is unread past the thread's anchor; the channel - // timeline was never marked. - let after = reply(&w.peer, 6, vec![w.mention()]).await; - let mut thread = item(&root, 1, &after); - thread["read_through_id"] = json!(anchor.id.to_hex()); - assert_eq!(w.counts(c).await, counts(true, 0, vec![thread])); - assert_eq!(w.row(c).await.read_through_id, None); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn a_deleted_message_is_no_witness_and_a_surviving_reply_still_is() { - let w = World::new().await; - let c = w.channel().await; - let root = w.post(c, &w.peer, w.now - 40 * DAY, vec![]).await; - let reply = |author, parent, at| w.reply(c, author, &root, Some(parent), w.now + at, vec![]); - - // The actor wrote the parent and never replied under it. - let wrote = reply(&w.actor, &root, 0).await; - reply(&w.peer, &wrote, 10).await; - // The actor's only reply to a peer's parent. - let once = reply(&w.peer, &root, 0).await; - let only = reply(&w.actor, &once, 1).await; - reply(&w.peer, &once, 11).await; - // Two replies by the actor to a peer's parent. - let twice = reply(&w.peer, &root, 0).await; - let first = reply(&w.actor, &twice, 1).await; - reply(&w.actor, &twice, 2).await; - let to_twice = reply(&w.peer, &twice, 12).await; - // The actor wrote the parent and also replied under it. - let both = reply(&w.actor, &root, 0).await; - let under = reply(&w.actor, &both, 1).await; - let to_both = reply(&w.peer, &both, 13).await; - - // How many replies the thread counts, the latest of them `to_both` until - // it stops counting. - let w = &w; - let thread = |mentions: u32| async move { - let row = w.row(c).await; - assert!(!row.unread, "the root predates the horizon"); - assert_eq!(row.threads.len(), 1); - assert_eq!(row.threads[0].mentions, mentions); - row.threads[0].latest_id.clone() - }; - // The four peer replies to the actor's conversations count. So do `once` - // and `twice`: peer replies to the root, which the actor has replied to - // (`wrote`, `both`). - assert_eq!(thread(6).await, to_both.id.to_hex()); - - // Each deletion changes only its own parent's conversation. - w.delete(&wrote).await; - assert_eq!(thread(5).await, to_both.id.to_hex()); - w.delete(&only).await; - assert_eq!(thread(4).await, to_both.id.to_hex()); - w.delete(&first).await; - assert_eq!(thread(4).await, to_both.id.to_hex()); - // With `wrote` and `both` gone the actor has no reply to the root either. - w.delete(&both).await; - assert_eq!(thread(2).await, to_both.id.to_hex()); - w.delete(&under).await; - assert_eq!(thread(1).await, to_twice.id.to_hex()); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn membership_is_scoped_to_the_replys_own_channel() { - let w = World::new().await; - let (a, b) = (w.channel().await, w.channel().await); - let old = w.now - 40 * DAY; - - // The actor replied to a peer's parent, but that reply is stored in B. - let theirs = w.post(a, &w.peer, old, vec![]).await; - w.reply(b, &w.actor, &theirs, Some(&theirs), old, vec![]) - .await; - let in_a = w - .reply(a, &w.peer, &theirs, Some(&theirs), w.now, vec![]) - .await; - assert_eq!(w.counts(a).await, counts(false, 0, vec![])); - - // The actor wrote a parent in A. A peer reply in B claims it as its parent. - let mine = w.post(a, &w.actor, old, vec![]).await; - w.reply(b, &w.peer, &mine, Some(&mine), w.now, vec![]).await; - assert_eq!(w.counts(b).await, counts(false, 0, vec![])); - - // The same parents count once the actor is a member in the reply's channel. - w.reply(a, &w.actor, &theirs, Some(&theirs), old, vec![]) - .await; - let to_mine = w.reply(a, &w.peer, &mine, Some(&mine), w.now, vec![]).await; - // Equal reply times: root ID order. - let mut threads = [item(&theirs, 1, &in_a), item(&mine, 1, &to_mine)]; - threads.sort_by(|x, y| x["root_id"].as_str().cmp(&y["root_id"].as_str())); - assert_eq!(w.counts(a).await, counts(false, 0, threads.to_vec())); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn membership_is_exact_behind_a_busy_parent() { - let w = World::new().await; - let c = w.channel().await; - let old = w.now - 40 * DAY; - let root = w.post(c, &w.peer, old, vec![]).await; - // The actor's reply is the oldest of 259 under one parent: a 256-row - // window over the parent's replies would never reach it. - w.reply(c, &w.actor, &root, Some(&root), old, vec![]).await; - let mut last = root.clone(); - for i in 0..258 { - last = w - .reply(c, &w.peer, &root, Some(&root), w.now + i, vec![]) - .await; - } - assert_eq!( - w.counts(c).await, - counts(false, 0, vec![item(&root, 258, &last)]) - ); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn threads_are_capped_after_replies_that_do_not_count_are_removed() { - let w = World::new().await; - let c = w.channel().await; - let old = w.now - 40 * DAY; - // The actor's thread has the oldest unread reply. A cap's worth of - // unjoined threads are newer and would fill the list if the cap came - // first. - let mine = w.post(c, &w.actor, old, vec![]).await; - let answer = w.reply(c, &w.peer, &mine, Some(&mine), w.now, vec![]).await; - for i in 1..=MAX_THREAD_SUMMARIES as u64 { - let root = w.post(c, &w.peer, old, vec![]).await; - w.reply(c, &w.peer, &root, Some(&root), w.now + i, vec![]) - .await; - } - assert_eq!( - w.counts(c).await, - counts(false, 0, vec![item(&mine, 1, &answer)]) - ); -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn the_parent_budget_leaves_undecided_replies_out() { - let w = World::new().await; - let c = w.channel().await; - // The actor is in the first conversation. Whether it is inside the budget - // depends on ID order, so only the bounds are asserted. - let mut mine = None; - for i in 0..1025 { - let author = if i == 0 { &w.actor } else { &w.peer }; - let root = w.post(c, author, w.now, vec![]).await; - let reply = w - .reply(c, &w.peer, &root, Some(&root), w.now + 1, vec![]) - .await; - if i == 0 { - mine = Some(item(&root, 1, &reply)); - } - // 1024 parents fit: 1023 peer roots, and the reply to the actor's. - // The independent SQL deadline may still withhold the answer. Past - // 1024, one reply is undecided. An undecided reply is left out. - if i >= 1023 { - let counted = w.counts(c).await; - assert!( - [vec![], vec![mine.clone().unwrap()]] - .map(|threads| counts(true, 0, threads)) - .contains(&counted), - "{i}: {counted}" - ); - } - } -} - -#[tokio::test] -#[ignore = "requires Postgres"] -async fn a_lookup_timeout_decides_nothing_and_preserves_the_callers_transaction() { - let w = World::new().await; - let c = w.channel().await; - let root = w.post(c, &w.peer, w.now, vec![]).await; - w.reply(c, &w.peer, &root, Some(&root), w.now + 1, vec![]) - .await; - let mut held = w.pool.begin().await.unwrap(); - // Force the real resolver to time out, then check that its outer snapshot - // and original statement budget remain usable. - sqlx::query("LOCK TABLE thread_metadata IN ACCESS EXCLUSIVE MODE") - .execute(&mut *held) - .await - .unwrap(); - let mut reader = w.pool.begin().await.unwrap(); - sqlx::query("SET LOCAL statement_timeout='2000ms'") - .execute(&mut *reader) - .await - .unwrap(); - for lock_timeout in ["0", "10ms"] { - sqlx::query("SELECT set_config('lock_timeout',$1,true)") - .bind(lock_timeout) - .execute(&mut *reader) - .await - .unwrap(); - sqlx::query("SET LOCAL jit=on") - .execute(&mut *reader) - .await - .unwrap(); - let result = participation::resolve( - &mut reader, - w.community, - &w.actor.public_key().to_bytes(), - &[(c, root.id.as_bytes().to_vec())], - ) - .await - .unwrap(); - assert!(result.is_empty(), "timeout provides no negative evidence"); - let setting: String = sqlx::query_scalar("SHOW statement_timeout") - .fetch_one(&mut *reader) - .await - .unwrap(); - assert_eq!(setting, "2s"); - let jit: String = sqlx::query_scalar("SHOW jit") - .fetch_one(&mut *reader) - .await - .unwrap(); - assert_eq!(jit, "on", "optional inference restores caller settings"); - } - reader.rollback().await.unwrap(); - held.rollback().await.unwrap(); - assert_eq!(w.counts(c).await, counts(true, 0, vec![])); -} 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 136eb89148d..ad850d219c6 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -51,7 +51,8 @@ pub(super) async fn start_before_everything( ) { sqlx::query( "INSERT INTO personal_read_accounts (community_id,actor,started_at) - VALUES ($1,$2,'1900-01-01T00:00:00Z')", + VALUES ($1,$2,'1900-01-01T00:00:00Z') + ON CONFLICT (community_id,actor) DO UPDATE SET started_at=excluded.started_at", ) .bind(community.as_uuid()) .bind(actor.to_bytes().as_slice()) @@ -60,14 +61,55 @@ pub(super) async fn start_before_everything( .unwrap(); } +/// Store `event` as a canonical reply in `root`'s thread, through ingest. +pub(super) async fn insert_reply( + db: &Db, + community: CommunityId, + channel: Uuid, + root: &nostr::Event, + event: &nostr::Event, +) { + let at = |e: &nostr::Event| { + chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap() + }; + db.insert_event_with_thread_metadata( + community, + event, + Some(channel), + Some(crate::event::ThreadMetadataParams { + event_id: event.id.as_bytes(), + event_created_at: at(event), + channel_id: channel, + parent_event_id: Some(root.id.as_bytes()), + parent_event_created_at: Some(at(root)), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(at(root)), + depth: 1, + broadcast: event.tags.iter().any(|t| { + let t = t.as_slice(); + t.len() >= 2 && t[0] == "broadcast" && t[1] == "1" + }), + }), + ) + .await + .unwrap(); +} + /// Move every community message past the default horizon by author time, the /// only clock the unread window reads. async fn expire(pool: &PgPool, community: CommunityId) { - sqlx::query("UPDATE events SET created_at=created_at-interval '31 days' WHERE community_id=$1") + for table in [ + "events SET created_at=created_at", + "event_mentions SET event_created_at=event_created_at", + ] { + sqlx::query(sqlx::AssertSqlSafe(format!( + "UPDATE {table}-interval '31 days' WHERE community_id=$1" + ))) .bind(community.as_uuid()) .execute(pool) .await .unwrap(); + } } async fn sidebar(db: &Db, community: CommunityId, actor: &Keys) -> SidebarPage { @@ -105,12 +147,17 @@ fn mark_thread(channel: Uuid, root: &nostr::Event, message: &nostr::Event) -> Re /// Address every stored community message to `actor`, so the channel's /// mention count is exactly its unread top-level count. async fn mention_everywhere(pool: &PgPool, community: CommunityId, actor: &Keys) { - sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1") - .bind(community.as_uuid()) - .bind(serde_json::json!([["p", actor.public_key().to_hex()]])) - .execute(pool) - .await - .unwrap(); + 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 + ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(actor.public_key().to_hex()) + .execute(pool) + .await + .unwrap(); } async fn add_history(db: &Db, community: CommunityId, channel: Uuid, count: usize) -> nostr::Event { @@ -134,7 +181,7 @@ async fn add_history(db: &Db, community: CommunityId, channel: Uuid, count: usiz #[ignore = "requires Postgres"] async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keeps_latest() { let (db, _pool, community, channel, actor, _) = fixture().await; - let last = add_history(&db, community, channel, MAX_UNREAD_SCAN + 44).await; + let last = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await; let outcome = db .apply_personal_read_intent( community, @@ -156,7 +203,7 @@ async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keep #[ignore = "requires Postgres"] async fn personal_read_sidebar_author_window_excludes_expired_and_late_old_messages() { let (db, pool, community, channel, actor, _) = fixture().await; - let latest = add_history(&db, community, channel, MAX_UNREAD_SCAN + 44).await; + let latest = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await; expire(&pool, community).await; let row = &sidebar(&db, community, &actor).await.channels[0]; assert!( @@ -183,34 +230,30 @@ async fn personal_read_sidebar_author_window_excludes_expired_and_late_old_messa #[tokio::test] #[ignore = "requires Postgres"] -async fn personal_read_sidebar_window_budget_counts_boundary_and_ineligible_tail() { +async fn personal_read_counts_stop_at_the_cap_after_eligibility() { let (db, pool, community, channel, actor, _) = fixture().await; - // The fixture contributes one event, so this is exactly the evidence budget. - add_history(&db, community, channel, MAX_UNREAD_SCAN - 1).await; + // The fixture contributes one message, so the count reaches the cap. + add_history(&db, community, channel, MAX_UNREAD_COUNT - 1).await; mention_everywhere(&pool, community, &actor).await; let mentions = |page: SidebarPage| page.channels[0].mentions; assert_eq!( mentions(sidebar(&db, community, &actor).await), - MAX_UNREAD_SCAN as u32 + MAX_UNREAD_COUNT as u32 ); - let overflow = EventBuilder::new(Kind::Custom(9), "one beyond the budget") - .sign_with_keys(&Keys::generate()) - .unwrap(); - db.insert_event(community, &overflow, Some(channel)) - .await - .unwrap(); + add_history(&db, community, channel, 50).await; mention_everywhere(&pool, community, &actor).await; - // The budget bounds the count; the unexamined message is left out. assert_eq!( mentions(sidebar(&db, community, &actor).await), - MAX_UNREAD_SCAN as u32 + MAX_UNREAD_COUNT as u32, + "a count of the cap means at least that many" ); - // Expire all but 601 messages. Of these, 300 are own and 300 deleted. - // Eligibility is downstream of the bounded unread window, not the old 256 cap. - sqlx::query("WITH ranked AS (SELECT created_at,id,row_number() OVER (ORDER BY created_at DESC,id) AS n FROM events WHERE community_id=$1 AND channel_id=$2) - UPDATE events e SET created_at=CASE WHEN r.n>601 THEN e.created_at-interval '31 days' ELSE e.created_at END, - pubkey=CASE WHEN r.n<=300 THEN $3 ELSE e.pubkey END, - deleted_at=CASE WHEN r.n>300 AND r.n<=600 THEN now() ELSE NULL END + // The last to arrive stays; of the 100 before it, 50 become own and 50 + // deleted; the rest expire. The cap applies after eligibility, so one + // remains. + sqlx::query("WITH ranked AS (SELECT created_at,id,row_number() OVER (ORDER BY received_at DESC,id) AS n FROM events WHERE community_id=$1 AND channel_id=$2) + UPDATE events e SET created_at=CASE WHEN r.n>101 THEN e.created_at-interval '31 days' ELSE e.created_at END, + pubkey=CASE WHEN r.n BETWEEN 2 AND 51 THEN $3 ELSE e.pubkey END, + deleted_at=CASE WHEN r.n BETWEEN 52 AND 101 THEN now() ELSE NULL END FROM ranked r WHERE e.community_id=$1 AND e.created_at=r.created_at AND e.id=r.id") .bind(community.as_uuid()).bind(channel).bind(actor.public_key().to_bytes().as_slice()).execute(&pool).await.unwrap(); let page = sidebar(&db, community, &actor).await; @@ -694,7 +737,7 @@ async fn personal_read_counts_only_arrivals_after_the_account_starts() { async fn personal_read_unstarted_account_is_caught_up_past_the_scan_cap() { let (db, pool, community, channel, actor, _) = fixture().await; unstart(&pool, community).await; - let last = add_history(&db, community, channel, MAX_UNREAD_SCAN + 44).await; + let last = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await; let page = sidebar(&db, community, &actor).await; let row = &page.channels[0]; // The start floor covers every arrival, examined by the scan or not. @@ -710,24 +753,18 @@ async fn personal_read_unstarted_account_is_caught_up_past_the_scan_cap() { async fn personal_read_channel_and_thread_never_inherit_each_other() { let (db, pool, community, channel, actor, root) = fixture().await; let base = root.created_at.as_secs(); - // Directed, so they count outside the actor's conversations. + // Mentioning the actor, who therefore follows the thread. let reply_at = |at: u64| { - let (db, pool) = (db.clone(), pool.clone()); + let db = db.clone(); let mention = Tag::public_key(actor.public_key()); - let root = root.id; + let root = root.clone(); async move { let reply = EventBuilder::new(Kind::Custom(9), "unseen thread reply") .tags([mention]) .custom_created_at(nostr::Timestamp::from(at)) .sign_with_keys(&Keys::generate()) .unwrap(); - db.insert_event(community, &reply, Some(channel)) - .await - .unwrap(); - sqlx::query("INSERT INTO thread_metadata (community_id,event_id,event_created_at,channel_id,root_event_id,parent_event_id,depth) - VALUES ($1,$2,to_timestamp($3),$4,$5,$5,1)") - .bind(community.as_uuid()).bind(reply.id.as_bytes().as_slice()).bind(at as f64) - .bind(channel).bind(root.as_bytes().as_slice()).execute(&pool).await.unwrap(); + insert_reply(&db, community, channel, &root, &reply).await; reply } }; @@ -812,10 +849,9 @@ async fn personal_read_channel_and_thread_never_inherit_each_other() { .fetch_all(&pool) .await .unwrap(); - assert_eq!( - roots, - vec![root.id.as_bytes().to_vec()], - "thread reading never creates a channel frontier" + assert!( + roots.is_empty(), + "reading a thread never follows it or creates a channel frontier" ); } diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index 38e6e026009..35204ba8ee4 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -1,20 +1,44 @@ -//! Bounded evidence projection. The cap limits evidence, not the definition of -//! unread: an unexamined tail goes uncounted, so counts may undercount. +//! Sidebar read state, counted forward from each read position. A caught-up +//! reader's work is a few index probes per channel and per followed thread; +//! a reader who is behind walks only what arrived after their position, and +//! counts stop at `MAX_UNREAD_COUNT`. -use super::{model::*, participation, writes}; +use super::{model::*, writes}; use buzz_core::CommunityId; use chrono::{DateTime, Utc}; -use serde_json::Value; use sqlx::{Acquire, PgConnection, Row}; -use std::cmp::Reverse; -use std::collections::{hash_map::Entry, HashMap}; use uuid::Uuid; use crate::{observability, Db, DbError, Result}; +/// Event `e` is eligible: a conversation kind, not deleted. +const ELIGIBLE: &str = "e.kind=ANY($5) AND e.deleted_at IS NULL"; + +/// Event `e`, with its `thread_metadata` row `tm`, is on the channel timeline: +/// not a reply, and not a NIP-10 reply whose ancestry was never recorded +/// (neither counted nor a timeline anchor, since a mark cannot validate it). +/// Parity-tested against the shared NIP-10 parser. +const TIMELINE: &str = r#"(tm.root_event_id IS NULL OR tm.root_event_id=e.id) + AND NOT (tm.root_event_id IS NULL AND jsonb_typeof(e.tags)='array' + AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag) + WHERE jsonb_typeof(tag)='array' AND tag->>0='e' AND tag->>3='reply' + AND (tag->>1) COLLATE "C" ~ '^[0123456789abcdefABCDEF]{64}$'))"#; + +/// What the budgeted timeline walks carry out of their `LIMIT`. +const COLUMNS: &str = + "e.id, e.pubkey, e.created_at, e.received_at, e.kind, e.deleted_at, e.tags, e.channel_id"; + +const EVENT_THREAD_ROW: &str = "LEFT JOIN thread_metadata tm ON tm.community_id=$1 + AND tm.channel_id=e.channel_id AND tm.event_created_at=e.created_at AND tm.event_id=e.id"; + +/// Replies `tm` in followed thread `tf`, with their events `e`. +const THREAD_REPLIES: &str = "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"; + impl Db { - /// Read a bounded joined roster from the writer. One SQL statement projects - /// channels, event evidence and read authority at a compatible MVCC cut. + /// Read a bounded joined roster from the writer in one read-only snapshot. /// Callers must recheck admission/resource access before releasing this data. pub async fn personal_read_sidebar( &self, @@ -70,7 +94,7 @@ impl Db { ) .await?; let mut tx = conn.begin().await?; - // The horizon and frontier evidence share one read-only cut. + // The horizon and every read position share one read-only cut. sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") .execute(&mut *tx) .await?; @@ -78,225 +102,176 @@ impl Db { sqlx::query("SET LOCAL jit = off").execute(&mut *tx).await?; let actor_bytes = actor.to_bytes(); let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?; - // The inner event LIMIT is deliberately before eligibility filtering. - // This bounds rows/joins even with long deleted or self-authored runs. - // Aggregate equivalent eligible evidence before transfer. Multiplicity - // preserves counts. - // Validate relevant tag parts before compacting directed/ancestry facts. - // PostgreSQL scalar "p" ->> 0 is "p": the type check must reject it too. - // Canonical timeline roots share an empty (present) root sentinel. - // Tags are bounded before transfer; oversized/corrupt evidence is - // skipped, never falsely top-level/unmentioned/read. - // Latest comes from the unread scan, so its ID arrives no earlier than - // anything counted; the shallow probe answers only when the horizon - // holds no top-level message. Thread replies are never the timeline's - // latest: a timeline mark cannot use one as its anchor. - let rows = sqlx::query( + // A position is its frontier floored at the account's start (infinity + // before the actor's first intent, so nothing counts). Positions are + // arrival times and the indexes are by author time, which the relay + // accepts within the skew of arrival, so a walk forward from a + // position starts the skew before it (and never before the horizon). + // + // Timeline, per channel: + // - latest_id: the newest timeline message by author time within the + // scan budget, then the last to arrive among the newest budget of + // events authored up to twice the skew before it, which holds the + // last arrival unless that many arrived out of order; + // - unread: latest is someone else's, inside the horizon, and arrived + // after the position (an own post moves the position past it); + // - mentions: those of them that tag the actor (event_mentions), or + // every one in a DM. + // Threads: each followed thread whose newest reply could have arrived + // after its position (one index-only probe of + // idx_thread_metadata_window), then its last arrival the same way as + // latest_id, and its unread replies by others. + let sql = format!( r#"WITH roster AS MATERIALIZED ( - SELECT c.id,c.channel_type::text AS channel_type + SELECT c.id, c.channel_type='dm' AS dm, + GREATEST(cf.through_timestamp, COALESCE($7::timestamptz, 'infinity')) AS position, + encode(cf.through_message_id,'hex') AS read_through_id FROM channel_members cm JOIN channels c ON c.community_id=cm.community_id AND c.id=cm.channel_id + LEFT JOIN personal_read_frontiers cf ON cf.community_id=$1 AND cf.actor=$2 + AND cf.channel_id=c.id AND cf.root_id=''::bytea WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL AND c.deleted_at IS NULL AND ($3::uuid IS NULL OR c.id>$3) - AND ($9::uuid[] IS NULL OR c.id=ANY($9)) - ORDER BY c.id LIMIT $4 + AND ($4::uuid[] IS NULL OR c.id=ANY($4)) + ORDER BY c.id LIMIT $6 ) - SELECT r.*, COALESCE(e.latest_id, latest.latest_id) AS latest_id, - encode(cf.through_message_id,'hex') AS read_through_id, - COALESCE(e.evidence,'[]'::jsonb) AS evidence FROM roster r + SELECT r.id, r.read_through_id, encode(l.id,'hex') AS latest_id, u.unread, + COALESCE(m.mentions, 0) AS mentions, COALESCE(th.items, '[]') AS threads + FROM roster r + CROSS JOIN LATERAL (SELECT GREATEST(r.position-$8, $9) AS walk_from) w LEFT JOIN LATERAL ( - WITH candidates AS MATERIALIZED ( - SELECT id,created_at,received_at,kind,deleted_at,tags FROM events - WHERE community_id=$1 AND channel_id=r.id - ORDER BY created_at DESC,id LIMIT $5 - ) - SELECT (array_agg(encode(c.id,'hex') ORDER BY c.received_at DESC,c.id) - FILTER (WHERE c.kind=ANY($6) AND c.deleted_at IS NULL - AND NOT COALESCE(tm.root_event_id<>c.id,false) - -- Same unresolved-reply test as `threaded` below. - AND NOT (tm.root_event_id IS NULL AND jsonb_typeof(c.tags)='array' - AND EXISTS (SELECT 1 FROM jsonb_array_elements(c.tags) t(tag) - WHERE jsonb_typeof(tag)='array' AND tag->>0='e' - AND tag->>3='reply' - AND (tag->>1) COLLATE "C" ~ '^[0123456789abcdefABCDEF]{64}$')) - ))[1] AS latest_id - FROM candidates c LEFT JOIN thread_metadata tm ON tm.community_id=$1 - AND tm.channel_id=r.id AND tm.event_created_at=c.created_at AND tm.event_id=c.id - ) latest ON true - LEFT JOIN personal_read_frontiers cf ON cf.community_id=$1 AND cf.actor=$2 - AND cf.channel_id=r.id AND cf.root_id=''::bytea + SELECT e.created_at FROM ( + SELECT {COLUMNS} FROM events e WHERE e.community_id=$1 AND e.channel_id=r.id + ORDER BY e.created_at DESC, e.id DESC LIMIT $10 + ) e {EVENT_THREAD_ROW} + WHERE {ELIGIBLE} AND {TIMELINE} + ORDER BY e.created_at DESC, e.id DESC LIMIT 1 + ) n ON true LEFT JOIN LATERAL ( - WITH candidates AS MATERIALIZED ( - SELECT id,pubkey,created_at,received_at,deleted_at,kind,tags - FROM events WHERE community_id=$1 AND channel_id=r.id AND created_at >= $7 - ORDER BY created_at DESC,id LIMIT $8 - ), threaded AS MATERIALIZED ( - SELECT e.*, tm.root_event_id AS root, tm.parent_event_id AS parent, - COALESCE(tm.root_event_id<>e.id,false) AS is_reply, - -- A NIP-10 reply marker without canonical ancestry: neither - -- counted nor a timeline anchor, since a mark cannot validate - -- it. Parity-tested against the shared NIP-10 parser. - tm.root_event_id IS NULL AND jsonb_typeof(e.tags)='array' - AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag) - WHERE jsonb_typeof(tag)='array' AND tag->>0='e' - AND tag->>3='reply' - AND (tag->>1) COLLATE "C" ~ '^[0123456789abcdefABCDEF]{64}$') - AS unresolved - FROM candidates e - LEFT JOIN thread_metadata tm ON tm.community_id=$1 AND tm.channel_id=r.id - AND tm.event_created_at=e.created_at AND tm.event_id=e.id - WHERE e.kind=ANY($6) AND e.deleted_at IS NULL - ), classified AS ( - SELECT e.*, - -- Every frontier is floored at the account's start. - COALESCE(e.received_at <= GREATEST(COALESCE($10::timestamptz,'infinity'), - CASE WHEN e.is_reply THEN tf.through_timestamp - ELSE cf.through_timestamp END),false) AS covered, - encode(tf.through_message_id,'hex') AS read_through_id - FROM threaded e - LEFT JOIN personal_read_frontiers tf ON tf.community_id=$1 AND tf.actor=$2 - AND tf.channel_id=r.id AND tf.root_id=e.root AND e.is_reply - WHERE e.pubkey<>$2 AND NOT e.unresolved - ), grouped AS ( - SELECT CASE WHEN root IS NULL THEN NULL - WHEN is_reply THEN encode(root,'hex') ELSE '' END AS root, - CASE WHEN is_reply THEN encode(parent,'hex') END AS parent, - is_reply, covered, read_through_id, - -- ->>0 also selects scalar "p"/"e": reject nonarrays first. - -- C collation matches Rust's ASCII case rules. - CASE WHEN octet_length(tags::text)<=8192 - AND jsonb_typeof(tags)='array' THEN - (SELECT CASE WHEN bool_or(jsonb_typeof(tag)<>'array' - OR jsonb_path_exists(tag,'strict $[*] ? (@.type() != "string")')) - THEN NULL ELSE jsonb_build_object( - 'directed',COALESCE(bool_or( - (tag->>0='p' AND lower((tag->>1) COLLATE "C")=encode($2,'hex')) - OR (tag->>0='broadcast' AND tag->>1='1')),false)) END - FROM jsonb_array_elements(tags) t(tag) - WHERE tag->>0 IN ('p','broadcast','e')) ELSE NULL END AS facts, - count(*) AS n, - (extract(epoch FROM max(received_at))*1000000)::bigint AS newest_arrival, - (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS newest_id, - (array_agg(extract(epoch FROM created_at)::bigint - ORDER BY received_at DESC,id))[1] AS newest_at - FROM classified - WHERE NOT covered OR root IS NULL - GROUP BY 1,2,3,4,5,6 - ) - SELECT (array_agg(encode(id,'hex') ORDER BY received_at DESC,id) - FILTER (WHERE NOT is_reply AND NOT unresolved))[1] AS latest_id, - (SELECT jsonb_agg(to_jsonb(grouped)) FROM grouped) AS evidence - FROM threaded - ) e ON true ORDER BY r.id"#, - ).bind(community.as_uuid()).bind(actor_bytes.as_slice()).bind(after) - .bind((limit+1) as i64).bind(MAX_CHANNEL_SCAN as i64) - .bind(ELIGIBLE_KINDS.as_slice()) - .bind(DateTime::from_timestamp_millis(account.cutoff_ms) - .ok_or_else(|| DbError::InvalidData("invalid unread cutoff".into()))?) - .bind(MAX_UNREAD_SCAN as i64) + SELECT e.id, e.pubkey, e.created_at, e.received_at FROM ( + SELECT {COLUMNS} FROM events e WHERE e.community_id=$1 AND e.channel_id=r.id + AND e.created_at BETWEEN n.created_at-2*$8 AND n.created_at + ORDER BY e.created_at DESC, e.id DESC LIMIT $10 + ) e {EVENT_THREAD_ROW} + WHERE {ELIGIBLE} AND {TIMELINE} + ORDER BY e.received_at DESC, e.id LIMIT 1 + ) l ON true + -- The last arrival is unread exactly when the timeline is: the + -- actor's own posts read through themselves. + CROSS JOIN LATERAL ( + SELECT COALESCE(l.received_at > r.position AND l.pubkey<>$2 + AND l.created_at >= $9, false) AS unread + ) u + LEFT JOIN LATERAL ( + SELECT count(*) AS mentions FROM ( + SELECT 1 FROM ( + SELECT {COLUMNS} FROM events e WHERE r.dm AND e.community_id=$1 + AND e.channel_id=r.id AND e.created_at >= w.walk_from + ORDER BY e.created_at, e.id LIMIT $10 + ) e {EVENT_THREAD_ROW} + WHERE e.received_at > r.position AND e.pubkey<>$2 AND {ELIGIBLE} AND {TIMELINE} + UNION ALL + SELECT 1 FROM event_mentions em JOIN events e ON e.community_id=$1 + AND e.created_at=em.event_created_at AND e.id=em.event_id {EVENT_THREAD_ROW} + WHERE NOT r.dm AND em.community_id=$1 AND em.pubkey_hex=$11 + AND em.channel_id=r.id AND em.event_created_at >= w.walk_from + AND e.channel_id=r.id AND e.received_at > r.position + AND e.pubkey<>$2 AND {ELIGIBLE} AND {TIMELINE} + LIMIT $12 + ) x WHERE u.unread + ) m ON true + LEFT JOIN LATERAL ( + SELECT jsonb_agg(jsonb_build_object('root_id',encode(t.root_id,'hex'), + 'unread',true,'mentions',t.n,'read_through_id',t.read_through_id, + 'latest_id',t.latest_id) ORDER BY t.latest_at DESC, t.root_id) AS items + FROM ( + SELECT tf.root_id, encode(tf.through_message_id,'hex') AS read_through_id, + encode(tl.id,'hex') AS latest_id, tl.created_at AS latest_at, tc.n + FROM personal_read_frontiers tf + CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, COALESCE($7, 'infinity')) AS position) p + CROSS JOIN LATERAL ( + SELECT tm.event_created_at AS created_at FROM thread_metadata tm + WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id + ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1 + ) tq + CROSS JOIN LATERAL ( + -- Eligibility is a per-row filter, not a join, so the + -- planner walks the window index newest-first and stops + -- at the first eligible reply instead of sorting them all. + SELECT tm.event_created_at AS created_at FROM thread_metadata tm + 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 (SELECT {ELIGIBLE} FROM events e WHERE e.community_id=$1 + AND e.created_at=tm.event_created_at AND e.id=tm.event_id) + ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1 + ) tn + CROSS JOIN LATERAL ( + SELECT e.id, e.created_at, e.received_at FROM ( + SELECT e.id, e.created_at, e.received_at FROM {THREAD_REPLIES} + AND tm.event_created_at BETWEEN tn.created_at-2*$8 AND tn.created_at + AND {ELIGIBLE} + ORDER BY tm.event_created_at DESC, tm.event_id LIMIT $10 + ) e + ORDER BY e.received_at DESC, e.id LIMIT 1 + ) tl + CROSS JOIN LATERAL ( + SELECT count(*) AS n FROM ( + SELECT 1 FROM {THREAD_REPLIES} + AND tm.event_created_at >= GREATEST(p.position-$8, $9) + AND e.received_at > p.position AND e.pubkey<>$2 AND {ELIGIBLE} + LIMIT $12 + ) x + ) tc + WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id + AND tf.root_id<>''::bytea + AND tq.created_at+$8 > p.position AND tl.received_at > p.position + AND tc.n > 0 + ORDER BY tl.created_at DESC, tf.root_id LIMIT $13 + ) t + ) th ON true + ORDER BY r.id"# + ); + let cutoff = DateTime::from_timestamp_millis(account.cutoff_ms) + .ok_or_else(|| DbError::InvalidData("invalid unread cutoff".into()))?; + // Only compile-time constants are interpolated. + let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(community.as_uuid()) + .bind(actor_bytes.as_slice()) + .bind(after) .bind(only) + .bind(ELIGIBLE_KINDS.as_slice()) + .bind((limit + 1) as i64) .bind(account.started_at) - .fetch_all(&mut *tx).await?; + .bind(skew()) + .bind(cutoff) + .bind(MAX_TIMELINE_SCAN as i64) + .bind(actor.to_hex()) + .bind(MAX_UNREAD_COUNT as i64) + .bind(MAX_THREAD_SUMMARIES as i64) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; let has_more = rows.len() > limit; - // An unstarted account has read every arrival, so nothing is unread. - let started = account.started_at.is_some(); - let mut channels = Vec::new(); - let mut pending = Vec::new(); + let mut channels = Vec::with_capacity(limit); for row in rows.into_iter().take(limit) { - let evidence: Value = row.try_get("evidence")?; - let evidence = evidence - .as_array() - .ok_or_else(|| DbError::InvalidData("invalid sidebar evidence".into()))?; - let evidence = if started { evidence.as_slice() } else { &[] }; - let channel_type: String = row.try_get("channel_type")?; - let mut unread = false; - let mut mentions = 0; - let mut threads: HashMap, Replies> = HashMap::new(); - let mut undirected = Vec::new(); - for e in evidence { - let n = e["n"] - .as_u64() - .filter(|n| *n <= MAX_UNREAD_SCAN as u64) - .ok_or_else(|| DbError::InvalidData("invalid evidence multiplicity".into()))? - as u32; - // Unusable tags prove nothing: the message is left out. - let Some(facts) = e["facts"].as_object() else { - continue; - }; - if e["covered"] == true { - continue; - } - let directed = - channel_type == "dm" || facts.get("directed") == Some(&Value::Bool(true)); - // Roots are timeline messages; descendants belong exclusively - // to their canonical thread. Never inherit the channel prefix. - if e["is_reply"] != true { - unread = true; - if directed { - mentions += n; - } - continue; - } - let Some(root) = e["root"].as_str().and_then(writes::event_id) else { - continue; - }; - let replies = Replies { - n, - newest: ( - e["newest_arrival"].as_i64().ok_or_else(invalid_newest)?, - e["newest_id"] - .as_str() - .ok_or_else(invalid_newest)? - .to_owned(), - ), - newest_at: e["newest_at"].as_i64().ok_or_else(invalid_newest)?, - read_through_id: e["read_through_id"].as_str().map(str::to_owned), - }; - // A directed reply counts whatever its conversation. Any other - // reply counts only in one of the actor's conversations. - if directed { - count(&mut threads, root, replies); - } else if let Some(parent) = e["parent"].as_str().and_then(writes::event_id) { - undirected.push((root, parent, replies)); - } - } - pending.push((threads, undirected)); + let threads: Vec = + serde_json::from_value(row.try_get("threads")?) + .map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?; + let mentions: i64 = row.try_get("mentions")?; channels.push(ChannelReadSummary { channel_id: row.try_get("id")?, - unread, - mentions, + unread: row.try_get("unread")?, + mentions: mentions as u32, read_through_id: row.try_get("read_through_id")?, latest_id: row.try_get("latest_id")?, - // Threads wait for conversation membership below. - threads: Vec::new(), + threads, }); } - let targets: Vec<_> = channels - .iter() - .zip(&pending) - .flat_map(|(channel, (_, undirected))| { - undirected - .iter() - .map(|(_, parent, _)| (channel.channel_id, parent.clone())) - }) - .collect(); - let members = participation::resolve(&mut tx, community, &actor_bytes, &targets).await?; - for (channel, (mut threads, undirected)) in channels.iter_mut().zip(pending) { - // Relevance is decided before the thread cap. An undecided - // membership leaves the reply out. - for (root, parent, replies) in undirected { - if members.get(&(channel.channel_id, parent)) == Some(&true) { - count(&mut threads, root, replies); - } - } - channel.threads = summarize(threads); - } let next_cursor = if has_more { channels.last().map(|c| c.channel_id) } else { None }; - tx.commit().await?; Ok(SidebarPage { channels, next_cursor, @@ -304,61 +279,8 @@ impl Db { } } -/// Uncovered replies in one canonical thread: one evidence group, or the -/// thread's counted total. -struct Replies { - n: u32, - /// Last of them to arrive: (arrival microseconds, lowercase hex ID). - newest: (i64, String), - /// Its author seconds. - newest_at: i64, - /// The thread's frontier anchor, shared by every group of the thread. - read_through_id: Option, -} - -/// Add replies that count to their thread. -fn count(threads: &mut HashMap, Replies>, root: Vec, replies: Replies) { - match threads.entry(root) { - Entry::Vacant(slot) => { - slot.insert(replies); - } - Entry::Occupied(mut slot) => { - let thread = slot.get_mut(); - thread.n += replies.n; - // Latest arrival first; equal arrivals break toward the smaller ID. - if (replies.newest.0, Reverse(&replies.newest.1)) - > (thread.newest.0, Reverse(&thread.newest.1)) - { - thread.newest = replies.newest; - thread.newest_at = replies.newest_at; - } - } - } -} - -fn invalid_newest() -> DbError { - DbError::InvalidData("invalid thread evidence".into()) -} - -/// Order newest unread reply first (root ID breaks ties) and cap the list. -fn summarize(threads: HashMap, Replies>) -> Vec { - let mut threads: Vec<_> = threads.into_iter().collect(); - threads.sort_unstable_by(|(a_root, a), (b_root, b)| { - b.newest_at - .cmp(&a.newest_at) - .then_with(|| a_root.cmp(b_root)) - }); - threads - .into_iter() - .take(MAX_THREAD_SUMMARIES) - .map(|(root, thread)| ThreadReadSummary { - root_id: hex::encode(root), - unread: true, - mentions: thread.n, - read_through_id: thread.read_through_id, - latest_id: thread.newest.1, - }) - .collect() +fn skew() -> chrono::Duration { + chrono::Duration::seconds(i64::from(MAX_ARRIVAL_SKEW_SECONDS)) } /// Read-time horizon and the account's start, in the caller's snapshot. @@ -384,37 +306,3 @@ pub(super) async fn read_account( started_at: row.try_get("started_at")?, }) } - -#[cfg(test)] -mod tests { - use super::*; - - fn replies(at: i64) -> Replies { - Replies { - n: 1, - newest: (at, String::new()), - newest_at: at, - read_through_id: None, - } - } - - #[test] - fn thread_rows_order_newest_first_break_ties_by_root_and_cap_at_twenty_five() { - // Literal contract boundaries deliberately do not derive from the constant. - for count in [0_u8, 1, 24, 25, 26, 27] { - // Pairwise-equal times exercise the tie-break. - let threads = (0..count) - .map(|i| (vec![i; 32], replies(i64::from(i / 2)))) - .collect(); - let roots: Vec<_> = summarize(threads).into_iter().map(|t| t.root_id).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(25) - .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 3dc5c131b36..f4fb0976776 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 @@ -1,4 +1,6 @@ use super::{postgres_tests::fixture, *}; +use crate::Db; +use nostr::{EventBuilder, Keys, Kind, Tag}; use serde_json::json; #[tokio::test] @@ -9,6 +11,16 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { let horizon = i64::from(DEFAULT_RETENTION_SECONDS) * 1000; // Addressed to the actor, so a counted message is also exactly one mention. let tags = json!([["p", actor.public_key().to_hex()]]); + 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", + ) + .bind(community.as_uuid()) + .bind(actor.public_key().to_hex()) + .execute(&pool) + .await + .unwrap(); for (kind, eligible_kind) in [ (9, true), (40002, true), @@ -32,6 +44,8 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { .bind(community.as_uuid()).bind(kind) .bind(if own { actor.public_key().to_bytes() } else { event.pubkey.to_bytes() }.as_slice()) .bind(deleted).bind(created as f64).bind(&tags).execute(&pool).await.unwrap(); + sqlx::query("UPDATE event_mentions SET event_created_at=to_timestamp($2::double precision/1000) WHERE community_id=$1") + .bind(community.as_uuid()).bind(created as f64).execute(&pool).await.unwrap(); let page = db .personal_read_sidebar( community, @@ -55,56 +69,6 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { } } -#[tokio::test] -#[ignore = "requires Postgres"] -async fn sidebar_compacted_tags_preserve_directed_and_corruption_rules() { - let (db, pool, community, _, actor, _) = fixture().await; - let actor_hex = actor.public_key().to_hex().to_uppercase(); - // Unusable tags prove nothing: that message is left out, neither unread - // nor a mention. - for (tags, unread, mentions) in [ - (json!([]), true, 0), - (json!(["p"]), false, 0), - (json!(["e"]), false, 0), - (json!(["broadcast"]), false, 0), - (json!([["p", actor_hex, "relay", "petname"]]), true, 1), - (json!([["p", "00".repeat(32)]]), true, 0), - (json!([["broadcast", "1", "extra"]]), true, 1), - (json!([["broadcast", "0"]]), true, 0), - (json!([["p", "00".repeat(32), 42]]), false, 0), - (json!([["broadcast", "0", 42]]), false, 0), - (json!([["e", "00".repeat(32), "", "root", 42]]), false, 0), - (json!([["p", actor_hex], ["p", "other", 42]]), false, 0), - (json!([["e", "00".repeat(32), "", "reply"]]), false, 0), - (json!([["x", 42]]), true, 0), - (json!({"p":actor_hex}), false, 0), - (json!([["p", "x".repeat(8193)]]), false, 0), - ] { - sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1") - .bind(community.as_uuid()) - .bind(&tags) - .execute(&pool) - .await - .unwrap(); - let page = db - .personal_read_sidebar( - community, - &actor.public_key(), - DEFAULT_RETENTION_SECONDS, - 20, - None, - ) - .await - .unwrap(); - let row = &page.channels[0]; - assert_eq!( - (row.unread, row.mentions), - (unread, mentions), - "tags={tags}" - ); - } -} - #[tokio::test] #[ignore = "requires Postgres"] async fn sidebar_ancestry_fact_matches_shared_nip10_parser() { @@ -169,7 +133,7 @@ async fn sidebar_ancestry_fact_matches_shared_nip10_parser() { #[tokio::test] #[ignore = "requires Postgres"] -async fn sidebar_directed_fact_follows_dm_mention_and_broadcast_rules() { +async fn sidebar_mentions_are_p_tags_in_a_stream_and_everything_in_a_dm() { let (db, pool, community, channel, actor, _) = fixture().await; let actor_hex = actor.public_key().to_hex(); let fullwidth: String = actor_hex @@ -177,54 +141,60 @@ async fn sidebar_directed_fact_follows_dm_mention_and_broadcast_rules() { .map(|c| if c.is_ascii_alphabetic() { 'Ａ' } else { c }) .collect(); assert_ne!(fullwidth, actor_hex); - // Whether each tag set is directed in a stream; in a DM every one is. + let tag = |parts: &[&str]| Tag::parse(parts.iter().copied()).unwrap(); + // Whether each message mentions the actor in a stream. A top-level + // broadcast is not a mention. let cases = [ - (json!([]), false), - (json!([["p", actor_hex]]), true), - (json!([["p", actor_hex.to_uppercase()]]), true), - (json!([["p", fullwidth]]), false), - (json!([["p"]]), false), - (json!([["broadcast", "1"]]), true), - (json!([["broadcast", "true"]]), false), - (json!([["p", "00".repeat(32)]]), false), + (vec![], false), + (vec![tag(&["p", &actor_hex])], true), + (vec![tag(&["p", &actor_hex.to_uppercase()])], true), + (vec![tag(&["p", &fullwidth])], false), + (vec![tag(&["broadcast", "1"])], false), + (vec![tag(&["p", &"00".repeat(32)])], false), ( - json!([["broadcast", "0"], ["p", actor_hex.to_uppercase()]]), + vec![tag(&["broadcast", "0"]), tag(&["p", &actor_hex])], true, ), ]; - for channel_type in ["stream", "dm"] { - sqlx::query( - "UPDATE channels SET channel_type=$3::channel_type WHERE community_id=$1 AND id=$2", - ) + let mut expected = 0; + for (tags, mentioned) in &cases { + let event = EventBuilder::new(Kind::Custom(9), "case") + .tags(tags.clone()) + .sign_with_keys(&Keys::generate()) + .unwrap(); + db.insert_event(community, &event, Some(channel)) + .await + .unwrap(); + expected += u32::from(*mentioned); + assert_eq!( + mentions(&db, community, &actor).await, + expected, + "tags={tags:?}" + ); + } + sqlx::query("UPDATE channels SET channel_type='dm' WHERE community_id=$1 AND id=$2") .bind(community.as_uuid()) .bind(channel) - .bind(channel_type) .execute(&pool) .await .unwrap(); - for (tags, in_stream) in &cases { - let expected = u32::from(channel_type == "dm" || *in_stream); - sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1") - .bind(community.as_uuid()) - .bind(tags) - .execute(&pool) - .await - .unwrap(); - let page = db - .personal_read_sidebar( - community, - &actor.public_key(), - DEFAULT_RETENTION_SECONDS, - 20, - None, - ) - .await - .unwrap(); - assert!(page.channels[0].unread); - assert_eq!( - page.channels[0].mentions, expected, - "channel_type={channel_type} tags={tags}" - ); - } - } + // Every message by someone else, the fixture's included. + assert_eq!( + mentions(&db, community, &actor).await, + cases.len() as u32 + 1 + ); +} + +async fn mentions(db: &Db, community: buzz_core::CommunityId, actor: &Keys) -> u32 { + db.personal_read_sidebar( + community, + &actor.public_key(), + DEFAULT_RETENTION_SECONDS, + 20, + None, + ) + .await + .unwrap() + .channels[0] + .mentions } 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 2bb86aab8da..b366279eea8 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 @@ -1,12 +1,14 @@ //! Thread summaries, anchor validation and targeted refresh. -use super::{postgres_tests::fixture, *}; +use super::{ + postgres_tests::{fixture, insert_reply}, + *, +}; use crate::{ channel::{ChannelType, ChannelVisibility}, Db, }; use buzz_core::CommunityId; use nostr::{EventBuilder, Keys, Kind, Tag}; -use sqlx::PgPool; use uuid::Uuid; async fn post( @@ -38,48 +40,41 @@ async fn post_as( event } -/// A canonical reply to the root: stored event plus its thread metadata. +/// A canonical reply to the root, through ingest. async fn reply( db: &Db, - pool: &PgPool, community: CommunityId, channel: Uuid, root: &nostr::Event, at: u64, tags: Vec, ) -> nostr::Event { - let event = post(db, community, channel, at, tags).await; - link(pool, community, channel, root, &event).await; - event + reply_as(db, community, channel, root, &Keys::generate(), at, tags).await } -async fn link( - pool: &PgPool, +async fn reply_as( + db: &Db, community: CommunityId, channel: Uuid, root: &nostr::Event, - event: &nostr::Event, -) { - sqlx::query("INSERT INTO thread_metadata (community_id,event_id,event_created_at,channel_id,root_event_id,parent_event_id,depth) - VALUES ($1,$2,to_timestamp($3),$4,$5,$5,1)") - .bind(community.as_uuid()).bind(event.id.as_bytes().as_slice()) - .bind(event.created_at.as_secs() as f64) - .bind(channel).bind(root.id.as_bytes().as_slice()).execute(pool).await.unwrap(); + author: &Keys, + at: u64, + tags: Vec, +) -> nostr::Event { + let event = EventBuilder::new(Kind::Custom(9), format!("reply at {at}")) + .tags(tags) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(author) + .unwrap(); + insert_reply(db, community, channel, root, &event).await; + event } -/// Put the actor in the root's conversation, so that plain replies to it -/// count. The actor's own reply is never unread. -async fn join( - db: &Db, - pool: &PgPool, - community: CommunityId, - channel: Uuid, - actor: &Keys, - root: &nostr::Event, -) { +/// The actor replies to the root, and so follows its thread. The actor's own +/// reply is never unread. +async fn join(db: &Db, community: CommunityId, channel: Uuid, actor: &Keys, root: &nostr::Event) { let at = root.created_at.as_secs(); - let own = post_as(db, community, channel, actor, at, vec![]).await; - link(pool, community, channel, root, &own).await; + reply_as(db, community, channel, root, actor, at, vec![]).await; } async fn sidebar(db: &Db, community: CommunityId, actor: &Keys) -> ChannelReadSummary { @@ -162,35 +157,6 @@ async fn mark_through_rejects_unresolved_missing_auxiliary_and_malformed_anchors assert_eq!(row.read_through_id, None, "nothing was marked"); } -#[tokio::test] -#[ignore = "requires Postgres"] -async fn channel_frontier_never_covers_an_epoch_zero_reply() { - let (db, pool, community, channel, actor, root) = fixture().await; - join(&db, &pool, community, channel, &actor, &root).await; - let ancient = reply(&db, &pool, community, channel, &root, 0, vec![]).await; - // A channel frontier exists, with no thread frontier: absent must not - // read as epoch zero or as the channel's. - apply( - &db, - community, - &actor, - channel_mark(channel, root.id.to_hex()), - ) - .await; - // Widen the horizon to reach the epoch, so that only a misread absent - // frontier could hide this reply. - let row = db - .personal_read_sidebar(community, &actor.public_key(), u32::MAX, 20, None) - .await - .unwrap() - .channels - .remove(0); - assert!(!row.unread); - assert_eq!(row.threads.len(), 1, "epoch-zero reply stays unread"); - assert_eq!(row.threads[0].mentions, 1); - assert_eq!(row.threads[0].latest_id, ancient.id.to_hex()); -} - #[tokio::test] #[ignore = "requires Postgres"] async fn thread_summaries_order_cap_and_anchor() { @@ -202,7 +168,7 @@ async fn thread_summaries_order_cap_and_anchor() { let mut roots = Vec::new(); for i in 0..n { let root = post(&db, community, channel, base + i, vec![]).await; - join(&db, &pool, community, channel, &actor, &root).await; + join(&db, community, channel, &actor, &root).await; roots.push(root); } let newest_root = roots.last().unwrap().clone(); @@ -220,11 +186,10 @@ async fn thread_summaries_order_cap_and_anchor() { let mut anchors = Vec::new(); for (i, root) in (0..).zip(&roots) { let at = base + 100 + offset(i); - anchors.push(reply(&db, &pool, community, channel, root, at, vec![]).await); + anchors.push(reply(&db, community, channel, root, at, vec![]).await); } reply( &db, - &pool, community, channel, &roots[2], @@ -234,7 +199,6 @@ async fn thread_summaries_order_cap_and_anchor() { .await; let twin = reply( &db, - &pool, community, channel, &roots[2], @@ -305,7 +269,7 @@ impl ThreadReadSummary { #[tokio::test] #[ignore = "requires Postgres"] async fn thread_on_a_never_unread_root_is_selectable_and_readable_by_itself() { - let (db, pool, community, channel, actor, root) = fixture().await; + let (db, _pool, community, channel, actor, root) = fixture().await; let base = root.created_at.as_secs(); let mention = || vec![Tag::parse(["p", &actor.public_key().to_hex()]).unwrap()]; // A diff is never unread, not even one addressed to the actor. @@ -317,8 +281,8 @@ async fn thread_on_a_never_unread_root_is_selectable_and_readable_by_itself() { db.insert_event(community, &diff, Some(channel)) .await .unwrap(); - let on_diff = reply(&db, &pool, community, channel, &diff, base + 10, mention()).await; - let elsewhere = reply(&db, &pool, community, channel, &root, base + 20, mention()).await; + let on_diff = reply(&db, community, channel, &diff, base + 10, mention()).await; + let elsewhere = reply(&db, community, channel, &root, base + 20, mention()).await; let row = sidebar(&db, community, &actor).await; // The undirected fixture root is unread; the directed diff is not counted. diff --git a/crates/buzz-db/src/store/personal_read/writes.rs b/crates/buzz-db/src/store/personal_read/writes.rs index 9d4c4a6586c..26736ff69a7 100644 --- a/crates/buzz-db/src/store/personal_read/writes.rs +++ b/crates/buzz-db/src/store/personal_read/writes.rs @@ -24,7 +24,8 @@ pub(super) async fn deadlines(conn: &mut PgConnection) -> Result<()> { } /// Start the account if this is the actor's first read intent, then serialize -/// private frontier writes, never shared conversation rows. +/// private frontier writes, never shared conversation rows. NO KEY UPDATE +/// leaves ingest's foreign-key checks (which create thread rows) unblocked. pub(super) async fn lock_account( conn: &mut PgConnection, community: CommunityId, @@ -40,7 +41,8 @@ pub(super) async fn lock_account( .execute(&mut *conn) .await?; sqlx::query( - "SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2 FOR UPDATE", + "SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2 + FOR NO KEY UPDATE", ) .bind(community.as_uuid()) .bind(actor) @@ -159,7 +161,8 @@ pub(super) async fn valid_target( } /// Advance a monotone frontier. The anchor ID follows the greatest arrival; -/// an equal arrival keeps the existing anchor. +/// an equal arrival keeps the existing anchor. Reading a thread never follows +/// it: only a followed thread has a row to advance. async fn frontier( conn: &mut PgConnection, community: CommunityId, @@ -168,7 +171,7 @@ async fn frontier( root: &[u8], msg: &Message, ) -> Result<()> { - sqlx::query( + let sql = if root.is_empty() { "INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id, through_timestamp, through_message_id) VALUES ($1,$2,$3,$4,$5,$6) @@ -176,10 +179,23 @@ async fn frontier( SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, EXCLUDED.through_timestamp), through_message_id=CASE WHEN EXCLUDED.through_timestamp > personal_read_frontiers.through_timestamp - THEN EXCLUDED.through_message_id ELSE personal_read_frontiers.through_message_id END", - ).bind(community.as_uuid()).bind(actor).bind(target.channel_id).bind(root) - .bind(msg.received_at).bind(&msg.id) - .execute(&mut *conn).await?; + THEN EXCLUDED.through_message_id ELSE personal_read_frontiers.through_message_id END" + } else { + "UPDATE personal_read_frontiers + SET through_timestamp=GREATEST(through_timestamp, $5), + through_message_id=CASE WHEN $5 > through_timestamp + THEN $6 ELSE through_message_id END + WHERE community_id=$1 AND actor=$2 AND channel_id=$3 AND root_id=$4" + }; + sqlx::query(sql) + .bind(community.as_uuid()) + .bind(actor) + .bind(target.channel_id) + .bind(root) + .bind(msg.received_at) + .bind(&msg.id) + .execute(&mut *conn) + .await?; Ok(()) } 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 63942726793..6d50c42815a 100644 --- a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs +++ b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs @@ -437,12 +437,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 gave the message's author an account; the actor has 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"); } diff --git a/docs/buzz-v1-read-state.md b/docs/buzz-v1-read-state.md index f89e2e07740..5e0aab9a2c6 100644 --- a/docs/buzz-v1-read-state.md +++ b/docs/buzz-v1-read-state.md @@ -75,9 +75,9 @@ marking one never moves another. | Field | Channel row | Thread row | |---|---|---| | `unread` | an unread top-level message counts | an unread reply counts | -| `mentions` | unread top-level messages directed at you | unread replies that count | +| `mentions` | unread top-level messages that tag you with `p`, or every unread one in a DM (at most 100) | unread replies (at most 100) | | `read_through_id` | the anchor of the timeline's frontier, or null | the anchor of the thread's frontier, or null | -| `latest_id` | the last eligible top-level message to arrive, or null | the last counted unread reply to arrive | +| `latest_id` | the last eligible top-level message to arrive, or null | the last reply to arrive | A thread reply never makes the channel row unread and never becomes its `latest_id`, including a reply carrying `broadcast=1`: it shows on its thread @@ -88,31 +88,41 @@ channel `latest_id` counts any author and ignores read progress. Null means the bounded scans found no top-level message, not that the channel is empty. `read_through_id` may name a message that has since been deleted. -`threads` lists threads with unread replies that count, newest unread reply -first by author time (then `root_id`), at most 25. Threads past the 25th are -omitted with their mentions, and nothing signals the omission: after marking -the listed threads read, a refresh can list the next ones. A thread row's -`unread` is always true in this version. No message bytes are included. +`threads` lists the threads you follow that have unread replies, newest +unread reply first by author time (then `root_id`), at most 25. Threads past +the 25th are omitted with their mentions, and nothing signals the omission: +after marking the listed threads read, a refresh can list the next ones. A +thread row's `unread` is always true in this version. No message bytes are +included. 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. -A top-level message is directed at you when its channel is a DM, it tags you -with `p`, or it carries `broadcast=1`. A reply counts only when it is directed -in the same way, or it is in one of your conversations: you wrote its direct -parent or have a reply to that same parent, in that channel. Conversation -membership uses only your 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. This is not Desktop notification -policy: follows and mutes do not affect these counts. - -Counts are what the bounded scans could establish, and may undercount. A message -whose tags are unusable, whose ancestry is unresolved, or whose conversation -membership is undecided is left out, as are messages beyond the scan bounds -(see [Bounds](#bounds-and-deployment)). +### Following + +Every top-level message counts on its channel row. A reply counts only in a +thread you follow. The relay records follows when it stores an eligible message, +in the same transaction: + +- Its author has read through it: the timeline for a top-level message, + otherwise the thread, which the author now follows. +- Each channel member it tags with `p` follows its thread (for a top-level + message, the thread rooted at it). +- A reply also makes the root's author follow, and in a DM every member. + +A new follower starts just before the message that made them follow, so it is +unread; one who already follows keeps their position. Reading never follows: +a thread `mark_through` moves an existing follow and otherwise changes nothing. +There is no unfollow yet. `broadcast=1` does not make anyone follow, so a +broadcast reply counts only for followers. Follows start when the relay that +records them is deployed; there is no backfill. Desktop notification follows +and mutes do not affect these counts. + +Counts are what the bounded walks could establish, and may undercount. A +message whose ancestry is unresolved is left out, as are messages beyond the +bounds (see [Bounds](#bounds-and-deployment)). The unread horizon defaults to 30 days (`BUZZ_V1_RETENTION_SECONDS`) and is measured in author time (`created_at`). It filters unread and mentions, not @@ -186,11 +196,21 @@ microsecond resolution. Three limits follow, none of which strands a badge: skew. - Messages with the identical stamp are read together. -A channel's `latest_id` is the last top-level 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 every top-level message counted. When -that scan holds no top-level message, it is the last top-level message to -arrive among the channel's 256 most recent events. +Counting walks forward from a position. Positions are arrival times but the +event indexes are by author time, which the relay accepts within 15 minutes of +arrival, so a walk starts 15 minutes of author time before the position. A +message backdated further than that arrives unseen; Order above still holds for +what the walk finds. A channel's `latest_id` is found from the newest top-level +message by author time: it is the last top-level message to arrive among those +authored up to 30 minutes before that one, which includes the last arrival. So +marking through it reads every top-level message counted. A thread's +`latest_id` is found the same way among its replies. + +A channel row is unread exactly when its `latest_id` is someone else's, inside +the horizon, and arrived after the timeline's position. Nothing earlier needs +checking: posting reads the timeline through the post, so an own `latest_id` +means everything before it is read, and an unread one is the anchor that clears +the row. A row is never unread without an anchor to mark. There is no import of earlier client read state. The relay records the account's `started_at` at its first applied read intent and never moves it. @@ -204,16 +224,14 @@ device-local. - 20 sidebar rows and 100 intents per request; 25 thread rows per channel, so a POST's rows for 100 channels fit the response limit. - 64 KiB write body; 1 MiB serialized API response. -- 4096 raw events per channel inside the horizon, before eligibility. -- The latest fallback probes 256 events; a long ineligible or reply-only tail - can leave `latest_id` null. -- Tag documents over 8192 bytes or malformed relevant tags leave the message - out. 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 and left - out. +- Counts stop at 100 per channel and per thread. +- Each walk reads at most 4096 raw events, before eligibility. The `latest_id` + probe reads the newest 4096 per channel; a longer ineligible or reply-only + tail (reactions, edits, thread replies) leaves it null and the row read. The + last-arrival search reads the newest 4096 of the 30-minute window, so more + than that many backdated arrivals can hide one. +- A caught-up reader costs a few index probes per channel and per followed + thread; a reader who is behind walks only what arrived after their position. - DB statement/lock deadlines and HTTP read deadlines bound work; a write's intents and its row refresh share one eight-second deadline after admission. Limits are containment, not a production capacity claim. @@ -222,9 +240,13 @@ Apply migrations 0056 through 0058 (or the equivalent desired schema). 0056 creates two empty private tables and no index on `events`, 0057 adds the nullable `personal_read_accounts.started_at` column, and 0058 adds the nullable `personal_read_frontiers.through_message_id` column and drops -`threads_through_timestamp`. 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. +`threads_through_timestamp`. Timeline walks use the existing +`idx_events_community_channel_created`, thread walks +`idx_thread_metadata_window`, and mention counts +`idx_event_mentions_pubkey_created`. Ingest writes a few frontier rows per +eligible channel message (the author, and anyone it makes follow), whether or +not the API is enabled, so follows exist once it is; there are no stored unread +counters. Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and `/buzz/v1/me/read-state`, plus database pool/statement metrics. No payload, @@ -236,7 +258,7 @@ local seed is not a DAU/concurrency or p95/p99 production acceptance result. Within `/buzz/v1`, clients must ignore unknown response object fields. The shape is stable: existing fields, their types, status variants and what a `mark_through` does remain. What the counts include is not yet: which messages -are unread or directed, which threads are listed, and the bounds may change +are unread or mentions, which threads are listed, and the bounds may change while v1 is pre-release. A client therefore replaces its rows with each response rather than reconciling them with its own counts. Breaking shape changes require an explicitly negotiated contract or a new API version. @@ -253,8 +275,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 followed threads and posted-in channels, not +observed messages, and has no fixed context-count ceiling. Leaving/rejoining does not erase progress; revoked access hides it. Soft-deleted channels are inaccessible, while hard channel deletion cascades their frontiers.