From 08c5c9407a4b1c623c9581c7258899fe40fa2d9f Mon Sep 17 00:00:00 2001 From: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> Date: Fri, 9 Oct 2026 19:17:22 -0700 Subject: [PATCH] feat(read-state): count thread replies only in followed threads Ingest records follows in the message's transaction: the author reads through their own message, and mentioned channel members, the root's author and DM members follow the thread just before it. The sidebar then counts forward from each read position instead of resolving participation and walking 4096 events back per channel, and counts stop at 100. A thread mark_through moves an existing follow and never creates one. participation.rs is removed. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> --- crates/buzz-db/src/store/event.rs | 26 + .../personal_read/arrival_postgres_tests.rs | 136 +++-- .../src/store/personal_read/follows.rs | 113 ++++ .../personal_read/follows_postgres_tests.rs | 373 ++++++++++++ crates/buzz-db/src/store/personal_read/mod.rs | 10 +- .../buzz-db/src/store/personal_read/model.rs | 34 +- .../src/store/personal_read/participation.rs | 127 ----- .../participation_postgres_tests.rs | 537 ------------------ .../src/store/personal_read/postgres_tests.rs | 122 ++-- .../src/store/personal_read/projection.rs | 476 ++++++---------- .../projection_postgres_tests.rs | 152 ++--- .../personal_read/threads_postgres_tests.rs | 92 +-- .../buzz-db/src/store/personal_read/writes.rs | 32 +- .../src/api/buzz_v1/postgres_tests.rs | 15 +- docs/buzz-v1-read-state.md | 106 ++-- 15 files changed, 1065 insertions(+), 1286 deletions(-) create mode 100644 crates/buzz-db/src/store/personal_read/follows.rs create mode 100644 crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs delete mode 100644 crates/buzz-db/src/store/personal_read/participation.rs delete mode 100644 crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs 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.