diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index b6ead54e06..24830bbf11 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,7 +705,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 60); + assert_eq!(migrations.len(), 61); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); @@ -713,6 +713,11 @@ mod postgres_tests { .sql .as_str() .contains("ALTER TABLE push_leases DROP COLUMN app_profile")); + assert_eq!(migrations[60].version, 61); + assert!(migrations[60] + .sql + .as_str() + .contains("idx_events_read_timeline")); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql @@ -2253,6 +2258,16 @@ mod postgres_tests { .sql .as_str(); assert!(through_message.contains("DROP COLUMN threads_through_timestamp")); + // 0061 adds the sticky thread unfollow. + let unfollowed_column = "unfollowed boolean not null default false \ + check (not unfollowed or root_id <> ''::bytea), "; + assert!(MIGRATOR + .iter() + .find(|m| m.version == 61) + .expect("read-state access path migration") + .sql + .as_str() + .contains("$d$BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea)$d$")); for (table, definition) in personal.tables { let in_schema = schema.tables.get(&table).map(|schema_definition| { if table == "personal_read_accounts" { @@ -2262,7 +2277,11 @@ mod postgres_tests { } }); let definition = if table == "personal_read_frontiers" { - definition.replacen(threads_through_column, through_message_column, 1) + definition.replacen( + threads_through_column, + &format!("{through_message_column}{unfollowed_column}"), + 1, + ) } else { definition }; diff --git a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs index beda4b199f..eca8ea3765 100644 --- a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs @@ -270,9 +270,11 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() .await .unwrap(); - // Also bind the public migrator: the advisory schema/destruction lock, - // ledger write and post-migration catalog checks must work with ingestion. - let production_result = migration::run_migrations(&pool).await; + // Also bind the migrator: the advisory schema/destruction lock, ledger + // write and post-migration catalog checks must work with ingestion. 0061 + // adds a column to thread_metadata, which takes a brief exclusive lock + // bounded by its lock_timeout, so it runs after the writer below. + let production_result = migration::run_migrations_through(&pool, 60).await; sqlx::query(insert) .bind(community) .bind(channel) @@ -280,10 +282,6 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() .execute(&mut *witness) .await .unwrap(); - let version: i64 = sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations") - .fetch_one(&mut *witness) - .await - .unwrap(); let final_oid: i64 = sqlx::query_scalar("SELECT 'idx_thread_metadata_window'::regclass::oid::bigint") .fetch_one(&mut *witness) @@ -297,6 +295,11 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() .unwrap(); drop(witness); writer.rollback().await.unwrap(); + migration::run_migrations(&pool).await.unwrap(); + let version: i64 = sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations") + .fetch_one(&pool) + .await + .unwrap(); drop_scratch_db(&admin, pool, &name).await; assert!( diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 9eec332b44..a217251d4e 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -447,15 +447,9 @@ 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?; + if let Some((channel, place)) = crate::personal_read::place(event, channel_id, None) { + crate::personal_read::record_message(tx, event, channel, place, result.0.received_at) + .await?; } crate::operator_listener::enqueue_mentions_in_transaction(tx, event).await?; } @@ -492,8 +486,8 @@ async fn insert_event_on( let not_before = extract_not_before(event); let result = sqlx::query( r#" - INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before, read_timeline) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT DO NOTHING "#, ) @@ -509,6 +503,7 @@ async fn insert_event_on( .bind(channel_id) .bind(d_tag.as_deref()) .bind(not_before) + .bind(crate::personal_read::place(event, channel_id, None).is_some_and(|(_, p)| p.on_timeline())) .execute(tx.conn()) .await?; @@ -1740,11 +1735,22 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( let received_at = Utc::now(); let d_tag = extract_d_tag(event); let not_before = extract_not_before(event); + let place = match &thread_meta { + Some(meta) => crate::personal_read::place( + event, + Some(meta.channel_id), + meta.parent_event_id.map(|parent| match meta.root_event_id { + Some(root) => (root, meta.root_event_created_at, meta.broadcast), + None => (parent, meta.parent_event_created_at, meta.broadcast), + }), + ), + None => crate::personal_read::place(event, channel_id, None), + }; let result = sqlx::query( r#" - INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before, read_timeline) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT DO NOTHING "#, ) @@ -1760,6 +1766,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( .bind(channel_id) .bind(d_tag.as_deref()) .bind(not_before) + .bind(place.is_some_and(|(_, p)| p.on_timeline())) .execute(tx.conn()) .await?; @@ -1774,6 +1781,12 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( ) .await?; + // Before the thread counters below: their root row is shared by every + // reply, so its lock must not be held across these writes. + if let Some((channel, place)) = place { + crate::personal_read::record_message(tx, event, channel, place, received_at).await?; + } + if let Some(ref meta) = thread_meta { let broadcast_val: bool = meta.broadcast; @@ -1783,8 +1796,8 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( (community_id, event_created_at, event_id, channel_id, parent_event_id, parent_event_created_at, root_event_id, root_event_created_at, - depth, broadcast) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + depth, broadcast, received_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING "#, ) @@ -1798,6 +1811,8 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( .bind(meta.root_event_created_at) .bind(meta.depth) .bind(broadcast_val) + // An eligible reply's arrival, for the read-state thread index. + .bind(place.and(meta.parent_event_id).map(|_| received_at)) .execute(tx.conn()) .await?; @@ -1882,22 +1897,6 @@ 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 f8add017bf..d0bb2ca900 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,7 +1,7 @@ //! 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, insert_reply}, + postgres_tests::{fixture, insert_reply, sync_reply_arrivals}, *, }; use crate::Db; @@ -23,6 +23,7 @@ async fn arrive(pool: &PgPool, community: CommunityId, event: &nostr::Event, arr .unwrap() .rows_affected(); assert_eq!(updated, 1, "arrival must land on exactly one stored event"); + sync_reply_arrivals(pool).await; } /// A message authored at `authored` that arrives at `arrived`. It mentions the @@ -254,10 +255,9 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() { assert!(row.threads.is_empty()); } -/// 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. +/// 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, @@ -310,37 +310,6 @@ async fn mark_as_read_clears_a_late_arrival_behind_reactions() { assert_eq!(counts(&db, community, &actor).await, (false, 0)); } -/// 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 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") - .bind(community.as_uuid()) - .bind(first.id.as_bytes().as_slice()) - .execute(&pool) - .await - .unwrap() - .rows_affected(); - assert_eq!(demoted, 1); - 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, None); - assert_eq!((row.unread, row.mentions), (false, 0)); -} - /// Store `event` as having arrived at exactly `seconds` plus `micros`. Built /// from integers, never a float, so microsecond order cannot hinge on rounding. async fn arrive_exact( @@ -360,6 +329,7 @@ async fn arrive_exact( .unwrap() .rows_affected(); assert_eq!(updated, 1, "arrival must land on exactly one stored event"); + sync_reply_arrivals(pool).await; let stored: chrono::DateTime = sqlx::query_scalar("SELECT received_at FROM events WHERE community_id=$1 AND id=$2") .bind(community.as_uuid()) diff --git a/crates/buzz-db/src/store/personal_read/follows.rs b/crates/buzz-db/src/store/personal_read/follows.rs index 680ccfb9dc..6681582efb 100644 --- a/crates/buzz-db/src/store/personal_read/follows.rs +++ b/crates/buzz-db/src/store/personal_read/follows.rs @@ -1,5 +1,5 @@ -//! Read state that ingest writes. A thread frontier row exists only for a -//! thread its actor follows; only this module creates one. +//! Read state that ingest writes. A thread frontier row exists for a thread +//! its actor follows or has unfollowed; ingest and follow intents create one. use super::model::ELIGIBLE_KINDS; use crate::{AdmittedTx, Result}; use chrono::{DateTime, Utc}; @@ -11,19 +11,75 @@ use uuid::Uuid; pub(crate) enum Place<'a> { /// A top-level timeline message. TopLevel, - /// A reply in the thread rooted at `root`. - Reply { root: &'a [u8] }, + /// A reply in the thread rooted at `root`, authored at `root_created_at`; + /// a broadcast reply is also on the channel timeline. + Reply { + root: &'a [u8], + root_created_at: DateTime, + broadcast: bool, + }, +} + +impl Place<'_> { + /// Whether the message counts on its channel timeline (`read_timeline`). + pub(crate) fn on_timeline(&self) -> bool { + matches!( + self, + Place::TopLevel + | Place::Reply { + broadcast: true, + .. + } + ) + } +} + +/// A reply's recorded ancestry: its root, the root's author time, and whether +/// it is a broadcast. +pub(crate) type Ancestry<'a> = (&'a [u8], Option>, bool); + +/// Where an eligible channel message sits, from its recorded ancestry. None +/// for other events, and for a reply whose ancestry was never (fully) +/// recorded: it is on neither read path. +pub(crate) fn place<'a>( + event: &Event, + channel: Option, + reply: Option>, +) -> Option<(Uuid, Place<'a>)> { + let channel = channel?; + if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) { + return None; + } + match reply { + Some((root, root_created_at, broadcast)) => Some(( + channel, + Place::Reply { + root, + root_created_at: root_created_at?, + broadcast, + }, + )), + None if buzz_core::nip10::parse_thread_markers(&event.tags) + .resolve() + .is_some() => + { + None + } + None => Some((channel, Place::TopLevel)), + } } /// 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; +/// otherwise the thread, which the author then follows (and the timeline +/// too for a broadcast reply); /// - 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. +/// to them. A follower keeps an existing position. An unfollow sticks until +/// its actor posts in the thread or is mentioned there again. pub(crate) async fn record_message( tx: &mut AdmittedTx, event: &Event, @@ -41,40 +97,49 @@ pub(crate) async fn record_message( .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), + let (read, thread, reply, root_created_at, broadcast) = match place { + Place::TopLevel => (&[][..], id, false, None, false), + Place::Reply { + root, + root_created_at, + broadcast, + } => (root, root, true, Some(root_created_at), broadcast), }; // 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 + "WITH involved (actor, root_id, through, message, refollow) AS ( + SELECT $4::bytea, $5::bytea, $6::timestamptz, $3::bytea, true UNION ALL - SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL FROM channel_members cm + SELECT $4, ''::bytea, $6, $3, true WHERE $11 + UNION ALL + SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL, true 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 + SELECT e.pubkey, $7, $6-interval '1 microsecond', NULL, false FROM events e + WHERE $9 AND e.community_id=$1 AND e.created_at=$10 AND e.id=$7 + AND e.channel_id=$2 AND e.pubkey<>$4 UNION ALL - SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL + SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL, false 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. + -- An existing follower keeps their position, so is not touched; + -- nor is an unfollow, unless this message re-follows (the author + -- posting, or a mention). 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 + SELECT f.unfollowed 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 + WHERE i.message IS NOT NULL OR f.unfollowed IS NULL OR (i.refollow AND f.unfollowed) + ORDER BY i.actor, i.root_id, i.message IS NULL, NOT i.refollow ), accounts AS ( INSERT INTO personal_read_accounts (community_id, actor) SELECT $1, r.actor FROM (SELECT DISTINCT actor FROM rows) r @@ -85,6 +150,9 @@ pub(crate) async fn record_message( WHERE a.found IS NULL ORDER BY r.actor ON CONFLICT DO NOTHING RETURNING 1 ) + -- A re-follow starts just before this message, like a new follow, + -- unless the unfollow is later. A row that a concurrent message + -- created is only advanced when it carries an anchor. 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 @@ -95,8 +163,10 @@ pub(crate) async fn record_message( 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", + THEN excluded.through_message_id ELSE personal_read_frontiers.through_message_id END, + unfollowed=false + WHERE excluded.through_message_id IS NOT NULL OR (personal_read_frontiers.unfollowed + AND excluded.through_timestamp > personal_read_frontiers.through_timestamp)", ) .bind(community.as_uuid()) .bind(channel) @@ -107,6 +177,8 @@ pub(crate) async fn record_message( .bind(thread) .bind(&mentioned) .bind(reply) + .bind(root_created_at) + .bind(broadcast) .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 index 33a2e75929..f1928326fd 100644 --- a/crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs @@ -95,7 +95,7 @@ async fn follow( .unwrap() } -/// (root, unread replies) for each listed thread. +/// (root, mentions) for each listed thread. fn threads(row: &ChannelReadSummary) -> Vec<(String, u32)> { row.threads .iter() @@ -129,7 +129,8 @@ async fn replying_follows_the_thread_and_reads_through_the_reply() { 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)]); + // Unread, but not directed at the actor. + assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]); assert_eq!(summary.threads[0].latest_id, other.id.to_hex()); assert_eq!(summary.threads[0].read_through_id, Some(own.id.to_hex())); } @@ -148,9 +149,9 @@ async fn a_mention_follows_a_member_from_just_before_it() { 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. + // The mention and what came after it are unread; not the reply before it. let summary = row(&db, community, channel, &bee).await; - assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]); + assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]); assert_ne!(summary.threads[0].latest_id, earlier.id.to_hex()); assert_ne!(summary.threads[0].latest_id, mentioning.id.to_hex()); } @@ -175,7 +176,7 @@ async fn a_later_mention_keeps_the_followers_position() { Some(Some(own.id.to_hex())) ); let summary = row(&db, community, channel, &actor).await; - assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]); + assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]); } #[tokio::test] @@ -188,7 +189,7 @@ async fn the_root_author_follows_on_the_first_reply() { 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)]); + assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]); } #[tokio::test] @@ -202,7 +203,8 @@ async fn a_top_level_mention_follows_the_thread_rooted_at_it() { 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)]); + // The reply is unread but does not mention the actor. + assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]); } #[tokio::test] @@ -236,8 +238,9 @@ async fn an_unfollowed_thread_never_counts_and_marking_it_follows_nothing() { 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. + // Only the broadcast reply is also on the timeline. assert_eq!((summary.unread, summary.mentions), (true, 0)); + assert_eq!(summary.latest_id, Some(last.id.to_hex())); let outcome = db .apply_personal_read_intent( @@ -270,7 +273,7 @@ async fn thread_counts_stop_at_the_cap() { 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; + reply(&db, community, channel, &root, &other, mention(&actor)).await; } let summary = row(&db, community, channel, &actor).await; assert_eq!( @@ -355,7 +358,7 @@ async fn a_nested_reply_follows_the_root_thread() { ); 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!(threads(&summary), [(root.id.to_hex(), 0)]); assert_eq!(summary.threads[0].latest_id, nested.id.to_hex()); } @@ -371,3 +374,274 @@ async fn a_message_stored_without_thread_metadata_reads_the_timeline() { assert!(!summary.unread); assert_eq!(summary.read_through_id, Some(own.id.to_hex())); } + +fn follow_intent(channel: Uuid, root: Option<&nostr::Event>, follow: bool) -> ReadIntent { + let target = ReadTarget { + channel_id: channel, + root_id: root.map(|r| r.id.to_hex()), + }; + if follow { + ReadIntent::Follow { target } + } else { + ReadIntent::Unfollow { target } + } +} + +async fn intent( + db: &Db, + community: CommunityId, + actor: &Keys, + intent: ReadIntent, +) -> IntentOutcome { + db.apply_personal_read_intent(community, &actor.public_key(), &intent) + .await + .unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn following_starts_now_and_needs_a_thread() { + let (db, pool, community, channel, actor, root) = fixture().await; + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + assert_eq!( + intent(&db, community, &actor, follow_intent(channel, None, true)).await, + IntentOutcome::Invalid + ); + let missing = message(&Keys::generate(), vec![]); + assert_eq!( + intent( + &db, + community, + &actor, + follow_intent(channel, Some(&missing), true) + ) + .await, + IntentOutcome::Blocked + ); + assert_eq!( + intent( + &db, + community, + &actor, + follow_intent(channel, Some(&root), true) + ) + .await, + IntentOutcome::Applied + ); + assert_eq!(follow(&pool, community, &actor, &root).await, Some(None)); + // The reply before the follow is read; the next one is not. + assert!(row(&db, community, channel, &actor) + .await + .threads + .is_empty()); + let next = 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(), 0)]); + assert_eq!(summary.threads[0].latest_id, next.id.to_hex()); + // Following again keeps the position. + intent( + &db, + community, + &actor, + follow_intent(channel, Some(&root), true), + ) + .await; + assert_eq!( + threads(&row(&db, community, channel, &actor).await).len(), + 1 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_unfollow_sticks_until_a_mention_or_own_post() { + let (db, pool, community, channel, actor, _) = fixture().await; + // The actor's own root: its author follows on every reply. + let root = post(&db, community, channel, &actor, vec![]).await; + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + assert_eq!( + threads(&row(&db, community, channel, &actor).await).len(), + 1 + ); + assert_eq!( + intent( + &db, + community, + &actor, + follow_intent(channel, Some(&root), false) + ) + .await, + IntentOutcome::Applied + ); + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + assert!(row(&db, community, channel, &actor) + .await + .threads + .is_empty()); + + // A mention follows again, from just before it. + let mentioning = reply( + &db, + community, + channel, + &root, + &Keys::generate(), + mention(&actor), + ) + .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, mentioning.id.to_hex()); + assert_eq!(summary.threads[0].read_through_id, None); + + // So does posting, which reads through the post. + intent( + &db, + community, + &actor, + follow_intent(channel, Some(&root), false), + ) + .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())) + ); + reply(&db, community, channel, &root, &Keys::generate(), vec![]).await; + assert_eq!( + threads(&row(&db, community, channel, &actor).await).len(), + 1 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_broadcast_reply_is_a_timeline_message_and_anchor() { + let (db, _pool, community, channel, actor, root) = fixture().await; + apply_mark(&db, community, channel, &actor, &root).await; + let broadcast = vec![ + Tag::parse(["broadcast", "1"]).unwrap(), + Tag::public_key(actor.public_key()), + ]; + let last = reply(&db, community, channel, &root, &Keys::generate(), broadcast).await; + let summary = row(&db, community, channel, &actor).await; + assert_eq!((summary.unread, summary.mentions), (true, 1)); + assert_eq!(summary.latest_id, Some(last.id.to_hex())); + assert_eq!( + apply_mark(&db, community, channel, &actor, &last).await, + IntentOutcome::Applied + ); + assert!(!row(&db, community, channel, &actor).await.unread); + + // A broadcast by the reader reads the timeline through it, as a post does. + post(&db, community, channel, &Keys::generate(), vec![]).await; + let broadcast = vec![Tag::parse(["broadcast", "1"]).unwrap()]; + let own = reply(&db, community, channel, &root, &actor, broadcast).await; + let summary = row(&db, community, channel, &actor).await; + assert!(!summary.unread); + assert_eq!(summary.read_through_id, Some(own.id.to_hex())); +} + +async fn apply_mark( + db: &Db, + community: CommunityId, + channel: Uuid, + actor: &Keys, + message: &nostr::Event, +) -> IntentOutcome { + let mark = ReadIntent::MarkThrough { + target: ReadTarget { + channel_id: channel, + root_id: None, + }, + message_id: message.id.to_hex(), + }; + intent(db, community, actor, mark).await +} + +async fn threads_page( + db: &Db, + community: CommunityId, + actor: &Keys, + limit: usize, + after: Option<&str>, +) -> ThreadsPage { + let cursor = after.map(|c| ThreadsCursor::parse(c).unwrap()); + db.personal_read_threads( + community, + &actor.public_key(), + DEFAULT_RETENTION_SECONDS, + limit, + cursor.as_ref(), + ) + .await + .unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn the_threads_view_lists_follows_newest_arrival_first() { + let (db, pool, community, channel, actor, _) = fixture().await; + let mut roots = Vec::new(); + for _ in 0..3 { + let root = post(&db, community, channel, &Keys::generate(), vec![]).await; + reply(&db, community, channel, &root, &actor, vec![]).await; + roots.push(root); + } + // The first thread has news, so it comes first; the rest are read. + reply( + &db, + community, + channel, + &roots[0], + &Keys::generate(), + mention(&actor), + ) + .await; + // Neither an unfollowed thread nor one with no reply is listed. + let quiet = post(&db, community, channel, &Keys::generate(), mention(&actor)).await; + let left = post(&db, community, channel, &Keys::generate(), vec![]).await; + reply(&db, community, channel, &left, &actor, vec![]).await; + intent( + &db, + community, + &actor, + follow_intent(channel, Some(&left), false), + ) + .await; + assert_eq!(follow(&pool, community, &actor, &quiet).await, Some(None)); + + let first = threads_page(&db, community, &actor, 2, None).await; + let rows: Vec<_> = first + .threads + .iter() + .map(|t| (t.thread.root_id.clone(), t.thread.unread, t.thread.mentions)) + .collect(); + assert_eq!( + rows, + [ + (roots[0].id.to_hex(), true, 1), + (roots[2].id.to_hex(), false, 0) + ] + ); + assert!(first.threads.iter().all(|t| t.channel_id == channel)); + let rest = threads_page(&db, community, &actor, 2, first.next_cursor.as_deref()).await; + let rest: Vec<_> = rest + .threads + .iter() + .map(|t| t.thread.root_id.clone()) + .collect(); + assert_eq!(rest, [roots[1].id.to_hex()]); + + // Leaving the channel hides its threads. + sqlx::query("UPDATE channel_members SET removed_at=now() WHERE community_id=$1 AND pubkey=$2") + .bind(community.as_uuid()) + .bind(actor.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .unwrap(); + assert!(threads_page(&db, community, &actor, 2, None) + .await + .threads + .is_empty()); +} diff --git a/crates/buzz-db/src/store/personal_read/mod.rs b/crates/buzz-db/src/store/personal_read/mod.rs index 130b55d81c..6df99443fc 100644 --- a/crates/buzz-db/src/store/personal_read/mod.rs +++ b/crates/buzz-db/src/store/personal_read/mod.rs @@ -10,7 +10,7 @@ mod model; mod projection; mod writes; -pub(crate) use follows::{record_message, Place}; +pub(crate) use follows::{place, record_message}; pub use model::*; #[cfg(test)] diff --git a/crates/buzz-db/src/store/personal_read/model.rs b/crates/buzz-db/src/store/personal_read/model.rs index 1b276918f6..c6dd551bb1 100644 --- a/crates/buzz-db/src/store/personal_read/model.rs +++ b/crates/buzz-db/src/store/personal_read/model.rs @@ -28,6 +28,29 @@ pub enum ReadIntent { /// Fixed anchor; retry must not substitute the latest message. message_id: String, }, + /// Follow a thread. A new follow, or one after an unfollow, starts now; + /// an existing follow keeps its position. + Follow { + /// The thread; a channel-only target is invalid. + target: ReadTarget, + }, + /// Stop following a thread until the actor posts in it or is mentioned + /// there again. + Unfollow { + /// The thread; a channel-only target is invalid. + target: ReadTarget, + }, +} + +impl ReadIntent { + /// The intent's target. + pub fn target(&self) -> &ReadTarget { + match self { + ReadIntent::MarkThrough { target, .. } + | ReadIntent::Follow { target } + | ReadIntent::Unfollow { target } => target, + } + } } /// Outcome for one independent transaction, never acknowledged before commit. @@ -57,16 +80,11 @@ 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; -/// 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. +/// Read positions are arrival times and mentions are indexed by author time, +/// so a mention 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]; @@ -89,7 +107,7 @@ pub struct ChannelReadSummary { /// 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, - /// Followed threads with unread replies, newest unread reply first. + /// Followed threads with unread replies, newest arrival first. pub threads: Vec, } @@ -98,9 +116,10 @@ pub struct ChannelReadSummary { pub struct ThreadReadSummary { /// Canonical thread-root event ID. pub root_id: String, - /// Whether an unread reply that counts was found. + /// Whether the last reply to arrive is unread. pub unread: bool, - /// Unread replies by others in this followed thread. + /// Unread replies directed at the actor: every one in a DM, otherwise + /// those that tag the actor with `p`. pub mentions: u32, /// The anchor of the thread's frontier: the marked or own reply that /// arrived last. None for a follower who has read nothing yet. @@ -117,3 +136,43 @@ pub struct SidebarPage { /// Exclusive UUID roster cursor. None means this roster scan exhausted. pub next_cursor: Option, } + +/// Maximum rows in one threads page. +pub const MAX_THREADS: usize = 50; + +/// One followed thread in the threads view. +#[derive(Debug, Serialize)] +pub struct FollowedThread { + /// Joined channel UUID. + pub channel_id: Uuid, + /// The thread's read state; `unread` may be false here. + #[serde(flatten)] + pub thread: ThreadReadSummary, +} + +/// A page of followed threads, newest arrival first. +#[derive(Debug, Serialize)] +pub struct ThreadsPage { + /// Followed threads in joined channels. + pub threads: Vec, + /// Opaque exclusive cursor. None means this scan exhausted. + pub next_cursor: Option, +} + +/// Where a threads page resumes: after this arrival and root. +#[derive(Clone, Debug)] +pub struct ThreadsCursor { + pub(super) latest_at: DateTime, + pub(super) root_id: Vec, +} + +impl ThreadsCursor { + /// Parse a `next_cursor` value; None if it is not one. + pub fn parse(value: &str) -> Option { + let (micros, root) = value.split_once('.')?; + Some(Self { + latest_at: DateTime::from_timestamp_micros(micros.parse().ok()?)?, + root_id: super::writes::event_id(root)?, + }) + } +} 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 ad850d219c..53dc50ff10 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -62,6 +62,20 @@ pub(super) async fn start_before_everything( } /// Store `event` as a canonical reply in `root`'s thread, through ingest. +/// Copy rewritten event arrivals onto their replies' thread rows, as ingest +/// stores them. +pub(super) async fn sync_reply_arrivals(pool: &PgPool) { + sqlx::query( + "UPDATE thread_metadata tm SET received_at=e.received_at FROM events e + WHERE e.community_id=tm.community_id AND e.created_at=tm.event_created_at + AND e.id=tm.event_id AND tm.received_at IS NOT NULL + AND tm.received_at<>e.received_at", + ) + .execute(pool) + .await + .unwrap(); +} + pub(super) async fn insert_reply( db: &Db, community: CommunityId, @@ -179,9 +193,9 @@ async fn add_history(db: &Db, community: CommunityId, channel: Uuid, count: usiz #[tokio::test] #[ignore = "requires Postgres"] -async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keeps_latest() { +async fn personal_read_sidebar_marked_history_is_read_and_keeps_latest() { let (db, _pool, community, channel, actor, _) = fixture().await; - let last = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await; + let last = add_history(&db, community, channel, 300).await; let outcome = db .apply_personal_read_intent( community, @@ -191,7 +205,6 @@ async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keep .await .unwrap(); assert_eq!(outcome, IntentOutcome::Applied); - // The scan never reaches the oldest 45 messages; nothing is unread. let row = &sidebar(&db, community, &actor).await.channels[0]; assert!(!row.unread); assert_eq!(row.mentions, 0); @@ -203,7 +216,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_TIMELINE_SCAN + 44).await; + let latest = add_history(&db, community, channel, 300).await; expire(&pool, community).await; let row = &sidebar(&db, community, &actor).await.channels[0]; assert!( @@ -734,13 +747,13 @@ async fn personal_read_counts_only_arrivals_after_the_account_starts() { #[tokio::test] #[ignore = "requires Postgres"] -async fn personal_read_unstarted_account_is_caught_up_past_the_scan_cap() { +async fn personal_read_unstarted_account_is_caught_up() { let (db, pool, community, channel, actor, _) = fixture().await; unstart(&pool, community).await; - let last = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await; + let last = add_history(&db, community, channel, 300).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. + // The start floor covers every arrival. assert!(!row.unread); assert_eq!(row.mentions, 0); assert!(row.threads.is_empty()); diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index 35204ba8ee..a3bd0569dc 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -1,7 +1,7 @@ -//! 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`. +//! Sidebar and threads read state, counted forward from each read position. +//! Each channel timeline and each followed thread is one index probe for its +//! last arrival; counts walk only what arrived after the position and stop at +//! `MAX_UNREAD_COUNT`. use super::{model::*, writes}; use buzz_core::CommunityId; @@ -11,31 +11,37 @@ 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"; +/// The last eligible reply `tl` to arrive in thread `tf` (idx_thread_metadata_read_arrival). +/// Deletion is a per-row filter, not a join, so the walk stops at the first +/// living reply instead of sorting them all. +const LATEST_REPLY: &str = "SELECT tm.event_id AS id, tm.received_at, e.pubkey, e.created_at + FROM (SELECT tm.event_id, tm.event_created_at, tm.received_at FROM thread_metadata tm + WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id AND tm.received_at IS NOT NULL + AND tm.channel_id=tf.channel_id + AND (SELECT e.deleted_at IS NULL 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.received_at DESC, tm.event_id LIMIT 1) tm + JOIN events e ON e.community_id=$1 AND e.created_at=tm.event_created_at AND e.id=tm.event_id"; -/// 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"; +/// Unread replies in thread `t` at position `t.position` that are directed at +/// the actor: every one in a DM (`t.dm`), otherwise those that tag the actor. +/// Nothing is walked unless `t.unread`. +const THREAD_MENTIONS: &str = "SELECT count(*) AS n FROM ( + SELECT 1 FROM thread_metadata tm JOIN events e ON e.community_id=$1 + AND e.created_at=tm.event_created_at AND e.id=tm.event_id + WHERE t.unread AND t.dm AND tm.community_id=$1 AND tm.root_event_id=t.root_id + AND tm.received_at > t.position AND tm.channel_id=t.channel_id + AND e.deleted_at IS NULL AND e.pubkey<>$2 AND e.created_at >= $4 + UNION ALL + SELECT 1 FROM event_mentions em JOIN thread_metadata tm ON tm.community_id=$1 + AND tm.event_created_at=em.event_created_at AND tm.event_id=em.event_id + JOIN events e ON e.community_id=$1 AND e.created_at=em.event_created_at AND e.id=em.event_id + WHERE t.unread AND NOT t.dm AND em.community_id=$1 AND em.pubkey_hex=$5 AND em.channel_id=t.channel_id + AND em.event_created_at >= GREATEST(t.position-$6, $4) + AND tm.root_event_id=t.root_id AND tm.received_at > t.position + AND e.deleted_at IS NULL AND e.pubkey<>$2 + LIMIT $7 + ) x"; impl Db { /// Read a bounded joined roster from the writer in one read-only snapshot. @@ -103,131 +109,77 @@ impl Db { let actor_bytes = actor.to_bytes(); let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?; // 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). + // before the actor's first intent, so nothing counts). // // 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; + // - latest_id: the last timeline message to arrive (idx_events_read_timeline); // - 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. + // - mentions: unread timeline messages that tag the actor, found by + // author time (event_mentions) from the skew before the position, + // or every one in a DM. + // Threads: each followed thread whose last reply is unread the same + // way, then its mentions for the listed ones only. let sql = format!( r#"WITH roster AS MATERIALIZED ( SELECT c.id, c.channel_type='dm' AS dm, - GREATEST(cf.through_timestamp, COALESCE($7::timestamptz, 'infinity')) AS position, + GREATEST(cf.through_timestamp, COALESCE($3::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 ($4::uuid[] IS NULL OR c.id=ANY($4)) - ORDER BY c.id LIMIT $6 + AND c.deleted_at IS NULL AND ($8::uuid IS NULL OR c.id>$8) + AND ($9::uuid[] IS NULL OR c.id=ANY($9)) + ORDER BY c.id LIMIT $10 ) 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 ( - 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 ( - 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} + SELECT e.id, e.pubkey, e.created_at, e.received_at FROM events e + WHERE e.community_id=$1 AND e.channel_id=r.id AND e.read_timeline + AND e.deleted_at IS NULL 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 + AND l.created_at >= $4, 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} + SELECT 1 FROM events e WHERE r.dm AND e.community_id=$1 + AND e.channel_id=r.id AND e.read_timeline AND e.deleted_at IS NULL + AND e.received_at > r.position AND e.pubkey<>$2 AND e.created_at >= $4 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 + AND e.created_at=em.event_created_at AND e.id=em.event_id + WHERE NOT r.dm AND em.community_id=$1 AND em.pubkey_hex=$5 + AND em.channel_id=r.id AND em.event_created_at >= GREATEST(r.position-$6, $4) + AND e.read_timeline AND e.deleted_at IS NULL + AND e.received_at > r.position AND e.pubkey<>$2 + LIMIT $7 ) 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 + 'unread',true,'mentions',tc.n,'read_through_id',t.read_through_id, + 'latest_id',encode(t.latest_id,'hex')) + 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 + SELECT tf.root_id, tf.channel_id, r.dm, p.position, true AS unread, + encode(tf.through_message_id,'hex') AS read_through_id, + tl.id AS latest_id, tl.received_at AS latest_at 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 + CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, + COALESCE($3, 'infinity')) AS position) p + CROSS JOIN LATERAL ({LATEST_REPLY}) tl 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 + AND tf.root_id<>''::bytea AND NOT tf.unfollowed + AND tl.received_at > p.position AND tl.pubkey<>$2 AND tl.created_at >= $4 + ORDER BY tl.received_at DESC, tf.root_id LIMIT $11 + ) t CROSS JOIN LATERAL ({THREAD_MENTIONS}) tc ) th ON true ORDER BY r.id"# ); @@ -237,16 +189,14 @@ impl Db { let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(community.as_uuid()) .bind(actor_bytes.as_slice()) + .bind(account.started_at) + .bind(cutoff) + .bind(actor.to_hex()) + .bind(skew()) + .bind(MAX_UNREAD_COUNT as i64) .bind(after) .bind(only) - .bind(ELIGIBLE_KINDS.as_slice()) .bind((limit + 1) as i64) - .bind(account.started_at) - .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?; @@ -277,6 +227,106 @@ impl Db { next_cursor, }) } + + /// Followed threads in joined channels, newest arrival first, read or + /// not, from the writer in one read-only snapshot. A thread with no reply + /// yet is omitted. Callers must recheck admission/resource access before + /// releasing this data. + pub async fn personal_read_threads( + &self, + community: CommunityId, + actor: &nostr::PublicKey, + retention_seconds: u32, + limit: usize, + after: Option<&ThreadsCursor>, + ) -> Result { + if !(1..=MAX_THREADS).contains(&limit) { + return Err(DbError::InvalidData("invalid threads limit".into())); + } + let mut conn = observability::acquire_writer( + &self.pool, + observability::WriterOperation::SubscriptionHistory, + ) + .await?; + let mut tx = conn.begin().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") + .execute(&mut *tx) + .await?; + writes::deadlines(&mut tx).await?; + 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?; + // Every follow is probed for its last arrival and sorted: the cost + // grows with the actor's lifetime follows, not with community size. + let sql = format!( + r#"SELECT t.channel_id, encode(t.root_id,'hex') AS root_id, t.unread, + tc.n AS mentions, + encode(t.through_message_id,'hex') AS read_through_id, + encode(t.latest_id,'hex') AS latest_id, t.latest_at + FROM ( + SELECT tf.root_id, tf.channel_id, c.channel_type='dm' AS dm, p.position, + tf.through_message_id, tl.id AS latest_id, tl.received_at AS latest_at, + (tl.received_at > p.position AND tl.pubkey<>$2 AND tl.created_at >= $4) AS unread + FROM personal_read_frontiers tf + JOIN channel_members cm ON cm.community_id=$1 AND cm.channel_id=tf.channel_id + AND cm.pubkey=$2 AND cm.removed_at IS NULL + JOIN channels c ON c.community_id=$1 AND c.id=tf.channel_id AND c.deleted_at IS NULL + CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, + COALESCE($3::timestamptz, 'infinity')) AS position) p + CROSS JOIN LATERAL ({LATEST_REPLY}) tl + WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.root_id<>''::bytea + AND NOT tf.unfollowed AND ($8::timestamptz IS NULL OR tl.received_at < $8 + OR (tl.received_at=$8 AND tf.root_id>$9)) + ORDER BY tl.received_at DESC, tf.root_id LIMIT $10 + ) t CROSS JOIN LATERAL ({THREAD_MENTIONS}) tc + ORDER BY t.latest_at DESC, t.root_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(account.started_at) + .bind(cutoff) + .bind(actor.to_hex()) + .bind(skew()) + .bind(MAX_UNREAD_COUNT as i64) + .bind(after.map(|c| c.latest_at)) + .bind(after.map(|c| c.root_id.as_slice())) + .bind((limit + 1) as i64) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + let has_more = rows.len() > limit; + let mut threads = Vec::with_capacity(limit); + let mut last = None; + for row in rows.into_iter().take(limit) { + let mentions: i64 = row.try_get("mentions")?; + let root_id: String = row.try_get("root_id")?; + last = Some(( + row.try_get::, _>("latest_at")?, + root_id.clone(), + )); + threads.push(FollowedThread { + channel_id: row.try_get("channel_id")?, + thread: ThreadReadSummary { + root_id, + 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")?, + }, + }); + } + let next_cursor = last + .filter(|_| has_more) + .map(|(at, root)| format!("{}.{root}", at.timestamp_micros())); + Ok(ThreadsPage { + threads, + next_cursor, + }) + } } fn skew() -> chrono::Duration { 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 f4fb097677..77af8d257f 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,26 +1,16 @@ use super::{postgres_tests::fixture, *}; use crate::Db; use nostr::{EventBuilder, Keys, Kind, Tag}; -use serde_json::json; +/// Eligibility is classified at ingest (kind) and at read (author, deletion, +/// horizon). Each case is stored through ingest, then rewritten in place for +/// the read-time facts. #[tokio::test] #[ignore = "requires Postgres"] async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { - let (db, pool, community, _, actor, event) = fixture().await; + let (db, pool, community, channel, actor, _) = fixture().await; let now = chrono::Utc::now().timestamp_millis(); 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), @@ -28,9 +18,22 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { (45003, true), (1, false), (7, false), - (39002, false), (40008, false), ] { + sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .unwrap(); + // Addressed to the actor, so a counted message is also exactly one mention. + let author = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(kind), "case") + .tags([Tag::public_key(actor.public_key())]) + .sign_with_keys(&author) + .unwrap(); + db.insert_event(community, &event, Some(channel)) + .await + .unwrap(); for own in [false, true] { for deleted in [false, true] { // Now, then one minute inside and one minute outside the horizon. @@ -40,12 +43,13 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { (horizon + 60_000, false), ] { let created = now - age; - sqlx::query("UPDATE events SET kind=$2,pubkey=$3,deleted_at=CASE WHEN $4 THEN now() ELSE NULL END,created_at=to_timestamp($5::double precision/1000),tags=$6 WHERE community_id=$1") - .bind(community.as_uuid()).bind(kind) - .bind(if own { actor.public_key().to_bytes() } else { event.pubkey.to_bytes() }.as_slice()) - .bind(deleted).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(); + sqlx::query("UPDATE events SET pubkey=$3,deleted_at=CASE WHEN $4 THEN now() ELSE NULL END,created_at=to_timestamp($5::double precision/1000) WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()).bind(event.id.as_bytes().as_slice()) + .bind(if own { actor.public_key() } else { author.public_key() }.to_bytes().as_slice()) + .bind(deleted).bind(created as f64).execute(&pool).await.unwrap(); + sqlx::query("UPDATE event_mentions SET event_created_at=to_timestamp($3::double precision/1000) WHERE community_id=$1 AND event_id=$2") + .bind(community.as_uuid()).bind(event.id.as_bytes().as_slice()) + .bind(created as f64).execute(&pool).await.unwrap(); let page = db .personal_read_sidebar( community, @@ -69,10 +73,12 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() { } } +/// A message with a NIP-10 reply marker but no recorded ancestry is on +/// neither read path, by the shared parser's definition of a reply. #[tokio::test] #[ignore = "requires Postgres"] async fn sidebar_ancestry_fact_matches_shared_nip10_parser() { - let (db, pool, community, _, actor, _) = fixture().await; + let (db, pool, community, channel, actor, _) = fixture().await; let ids = [ "a".repeat(64), "A".repeat(64), @@ -104,12 +110,22 @@ async fn sidebar_ancestry_fact_matches_shared_nip10_parser() { buzz_core::nip10::parse_thread_markers_from_parts(tags.iter().map(Vec::as_slice)) .resolve() .is_some(); - sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1") + sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1") .bind(community.as_uuid()) - .bind(json!(tags)) .execute(&pool) .await .unwrap(); + let parsed: Vec = tags + .iter() + .map(|tag| Tag::parse(tag.iter().map(String::as_str)).unwrap()) + .collect(); + let event = EventBuilder::new(Kind::Custom(9), "case") + .tags(parsed) + .sign_with_keys(&Keys::generate()) + .unwrap(); + db.insert_event(community, &event, Some(channel)) + .await + .unwrap(); let page = db .personal_read_sidebar( community, 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 b366279eea..2259649b6b 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,6 +1,6 @@ //! Thread summaries, anchor validation and targeted refresh. use super::{ - postgres_tests::{fixture, insert_reply}, + postgres_tests::{fixture, insert_reply, sync_reply_arrivals}, *, }; use crate::{ @@ -179,7 +179,7 @@ async fn thread_summaries_order_cap_and_anchor() { channel_mark(channel, newest_root.id.to_hex()), ) .await; - // Threads 0 and 1 tie on newest reply time and the rest follow, newest + // Threads 0 and 1 tie on last arrival and the rest follow, newest // first. Thread 2 has one directed reply and two that arrive last, // together. let offset = |i: u64| if i < 2 { n * 10 } else { (n - i) * 10 }; @@ -206,14 +206,18 @@ async fn thread_summaries_order_cap_and_anchor() { vec![], ) .await; + // Replies arrive in author order, so the twin and thread 2's anchor + // arrive together. sqlx::query( - "UPDATE events SET received_at=(SELECT received_at FROM events WHERE id=$1) WHERE id=$2", + "UPDATE events e SET received_at=e.created_at FROM thread_metadata tm + WHERE tm.community_id=e.community_id AND tm.event_id=e.id + AND tm.received_at IS NOT NULL AND e.community_id=$1", ) - .bind(twin.id.as_bytes().as_slice()) - .bind(anchors[2].id.as_bytes().as_slice()) + .bind(community.as_uuid()) .execute(&pool) .await .unwrap(); + sync_reply_arrivals(&pool).await; let row = sidebar(&db, community, &actor).await; // Every root is read; the eight unread replies count only on threads. @@ -230,7 +234,7 @@ async fn thread_summaries_order_cap_and_anchor() { // The oldest unread thread is past the cap. assert_eq!(order(&row), [tied.to_vec(), ids(&roots[2..last])].concat()); let third = &row.threads[2]; - assert_eq!(third.mentions, 3); + assert_eq!(third.mentions, 1, "only the directed reply is a mention"); assert_eq!( third.latest_id, std::cmp::min(anchors[2].id.to_hex(), twin.id.to_hex()), @@ -249,8 +253,8 @@ async fn thread_summaries_order_cap_and_anchor() { [vec![tied[1].clone()], ids(&roots[2..])].concat() ); let mentions: Vec<_> = row.threads.iter().map(|t| t.mentions).collect(); - let mut expected = vec![1; MAX_THREAD_SUMMARIES]; - expected[1] = 3; + let mut expected = vec![0; MAX_THREAD_SUMMARIES]; + expected[1] = 1; assert_eq!(mentions, expected); } diff --git a/crates/buzz-db/src/store/personal_read/writes.rs b/crates/buzz-db/src/store/personal_read/writes.rs index 26736ff69a..8c15854799 100644 --- a/crates/buzz-db/src/store/personal_read/writes.rs +++ b/crates/buzz-db/src/store/personal_read/writes.rs @@ -88,6 +88,8 @@ struct Message { id: Vec, received_at: DateTime, root: Option>, + /// A reply that is also on the channel timeline. + broadcast: bool, } /// A channel event and its canonical ancestry. `kinds` bounds the lookup: an @@ -100,7 +102,7 @@ async fn message( kinds: Option<&[i32]>, ) -> Result> { let row = sqlx::query( - "SELECT e.id, e.received_at, e.tags, tm.root_event_id + "SELECT e.id, e.received_at, e.tags, tm.root_event_id, COALESCE(tm.broadcast, false) AS broadcast FROM events e LEFT JOIN thread_metadata tm ON tm.community_id=e.community_id AND tm.event_created_at=e.created_at AND tm.event_id=e.id AND tm.channel_id=e.channel_id WHERE e.community_id=$1 AND e.channel_id=$2 AND e.id=$3 @@ -125,6 +127,7 @@ async fn message( id, received_at, root, + broadcast: row.try_get("broadcast")?, }) }) .transpose() @@ -199,6 +202,38 @@ async fn frontier( Ok(()) } +/// Follow or unfollow a thread. A follow that changes state starts now, so +/// earlier replies are read; an unfollowed row keeps ingest from following +/// again, except as `record_message` allows. +async fn follow( + conn: &mut PgConnection, + community: CommunityId, + actor: &[u8], + channel: Uuid, + root: &[u8], + unfollow: bool, +) -> Result<()> { + sqlx::query( + "INSERT INTO personal_read_frontiers + (community_id, actor, channel_id, root_id, through_timestamp, unfollowed) + VALUES ($1,$2,$3,$4,now(),$5) + ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE + SET unfollowed=$5, + through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, now()), + through_message_id=CASE WHEN now() > personal_read_frontiers.through_timestamp + THEN NULL ELSE personal_read_frontiers.through_message_id END + WHERE personal_read_frontiers.unfollowed <> $5", + ) + .bind(community.as_uuid()) + .bind(actor) + .bind(channel) + .bind(root) + .bind(unfollow) + .execute(conn) + .await?; + Ok(()) +} + pub(super) async fn apply( conn: &mut PgConnection, community: CommunityId, @@ -225,13 +260,30 @@ pub(super) async fn apply( return Ok(IntentOutcome::Blocked); }; let is_reply = msg.root.as_ref().is_some_and(|r| r != &msg.id); - if (root.is_empty() && is_reply) + if (root.is_empty() && is_reply && !msg.broadcast) || (!root.is_empty() && msg.id != root && msg.root.as_ref() != Some(&root)) { return Ok(IntentOutcome::Blocked); } frontier(conn, community, actor, target, &root, &msg).await?; } + ReadIntent::Follow { target } | ReadIntent::Unfollow { target } => { + if target.root_id.is_none() { + return Ok(IntentOutcome::Invalid); + } + let Some(root) = valid_target(conn, community, actor, target).await? else { + return Ok(IntentOutcome::Blocked); + }; + follow( + conn, + community, + actor, + target.channel_id, + &root, + matches!(intent, ReadIntent::Unfollow { .. }), + ) + .await?; + } } Ok(IntentOutcome::Applied) } diff --git a/crates/buzz-relay/src/api/buzz_v1/handlers.rs b/crates/buzz-relay/src/api/buzz_v1/handlers.rs index f3dbf09565..8f7d1eb17c 100644 --- a/crates/buzz-relay/src/api/buzz_v1/handlers.rs +++ b/crates/buzz-relay/src/api/buzz_v1/handlers.rs @@ -7,7 +7,8 @@ use axum::{ response::Response, }; use buzz_db::personal_read::{ - ChannelReadSummary, ReadIntent, SidebarPage, MAX_CHANNELS, MAX_INTENTS, + ChannelReadSummary, ReadIntent, ThreadsCursor, ThreadsPage, MAX_CHANNELS, MAX_INTENTS, + MAX_THREADS, }; use serde::Deserialize; use serde_json::{json, Value}; @@ -74,27 +75,72 @@ pub(super) async fn sidebar( } } .map_err(|_| Error::unavailable())?; - auth::response(release(&state, &headers, &principal, page).await?) + let channels: Vec<_> = page.channels.iter().map(|c| c.channel_id).collect(); + release(&state, &headers, &principal, &channels).await?; + auth::response(page) + }) + .await + .map_err(|_| Error::unavailable())? +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct ThreadsQuery { + limit: Option, + cursor: Option, +} + +pub(super) async fn threads( + State(state): State>, + headers: HeaderMap, + OriginalUri(uri): OriginalUri, + query: Result, axum::extract::rejection::QueryRejection>, +) -> Result { + tokio::time::timeout(Duration::from_secs(8), async { + let principal = auth::authorize(&state, &headers, &uri, "GET", None).await?; + let Query(query) = query.map_err(|_| Error::invalid())?; + let limit = query.limit.unwrap_or(MAX_THREADS); + let cursor = query + .cursor + .map(|c| ThreadsCursor::parse(&c).ok_or_else(Error::invalid)) + .transpose()?; + if !(1..=MAX_THREADS).contains(&limit) { + return Err(Error::invalid()); + } + let page: ThreadsPage = state + .db + .personal_read_threads( + principal.tenant.community(), + &principal.actor, + state.config.buzz_v1_retention_seconds, + limit, + cursor.as_ref(), + ) + .await + .map_err(|_| Error::unavailable())?; + let channels: BTreeSet<_> = page.threads.iter().map(|t| t.channel_id).collect(); + let channels: Vec<_> = channels.into_iter().collect(); + release(&state, &headers, &principal, &channels).await?; + auth::response(page) }) .await .map_err(|_| Error::unavailable())? } /// Release projected rows only after admission and the actor's membership of -/// every row are rechecked outside the projection snapshot. +/// every row's channel are rechecked outside the projection snapshot. async fn release( state: &AppState, headers: &HeaderMap, principal: &auth::Principal, - page: SidebarPage, -) -> Result { + channels: &[Uuid], +) -> Result<(), Error> { auth::recheck(state, headers, principal).await?; - let channels: Vec<_> = page.channels.iter().map(|c| c.channel_id).collect(); let memberships = state .db .membership_pairs( principal.tenant.community(), - &channels, + channels, &[principal.actor.to_bytes().to_vec()], ) .await @@ -102,7 +148,7 @@ async fn release( if memberships.len() != channels.len() { return Err(Error::unavailable()); } - Ok(page) + Ok(()) } /// Updated rows for the channels a batch applied to. A channel the actor has @@ -126,7 +172,9 @@ async fn refreshed( ) .await .map_err(|_| Error::unavailable())?; - rows.extend(release(state, headers, principal, page).await?.channels); + let ids: Vec<_> = page.channels.iter().map(|c| c.channel_id).collect(); + release(state, headers, principal, &ids).await?; + rows.extend(page.channels); } Ok(rows) } @@ -168,8 +216,7 @@ pub(super) async fn write( .await .unwrap_or_else(|_| json!({"status":"unknown","retryable":true})); if outcome["status"] == "applied" { - let ReadIntent::MarkThrough { target, .. } = &intent; - applied.insert(target.channel_id); + applied.insert(intent.target().channel_id); } outcomes.push(outcome); } diff --git a/crates/buzz-relay/src/api/buzz_v1/mod.rs b/crates/buzz-relay/src/api/buzz_v1/mod.rs index d462ac15aa..0236dd7761 100644 --- a/crates/buzz-relay/src/api/buzz_v1/mod.rs +++ b/crates/buzz-relay/src/api/buzz_v1/mod.rs @@ -17,6 +17,7 @@ pub const BASE_PATH: &str = "/buzz/v1"; pub fn router(state: Arc) -> Router { Router::new() .route("/me/sidebar", get(handlers::sidebar)) + .route("/me/threads", get(handlers::threads)) .route("/me/read-state", post(handlers::write)) .fallback(|| async { auth::Error::new(axum::http::StatusCode::NOT_FOUND, "not_found") }) .with_state(state) 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 6d50c42815..c7becba628 100644 --- a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs +++ b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs @@ -361,6 +361,115 @@ async fn accessory_batch_outcomes_are_independent_and_return_applied_rows() { assert_ne!(response.status(), StatusCode::OK); } +#[tokio::test] +#[ignore = "requires Postgres"] +async fn accessory_follow_intents_and_threads_view() { + let fixture = crate::api::bridge::postgres_tests::bridge_handler_test_state() + .await + .unwrap(); + let mut state = (*fixture).clone(); + Arc::make_mut(&mut state.config).buzz_v1_enabled = true; + let state = Arc::new(state); + let host = format!("bff-threads-{}.local", uuid::Uuid::new_v4()); + let community = state + .db + .ensure_configured_community(&host) + .await + .unwrap() + .id; + let actor = Keys::generate(); + let channel = state + .db + .create_channel( + community, + "threads", + buzz_db::channel::ChannelType::Stream, + buzz_db::channel::ChannelVisibility::Open, + None, + &actor.public_key().to_bytes(), + None, + ) + .await + .unwrap() + .id; + start_before_everything(&state, community, &actor).await; + let mut roots = Vec::new(); + for _ in 0..2 { + let root = EventBuilder::new(Kind::Custom(9), "root") + .sign_with_keys(&Keys::generate()) + .unwrap(); + state + .db + .insert_event(community, &root, Some(channel)) + .await + .unwrap(); + roots.push(root); + } + let write_path = "/buzz/v1/me/read-state"; + let body = serde_json::to_vec(&json!({"intents":[ + {"type":"follow","target":{"channel_id":channel}}, + {"type":"follow","target":{"channel_id":channel,"root_id":roots[0].id.to_hex()}}, + {"type":"follow","target":{"channel_id":channel,"root_id":roots[1].id.to_hex()}}, + {"type":"unfollow","target":{"channel_id":channel,"root_id":roots[1].id.to_hex()}} + ]})) + .unwrap(); + let auth = proof(&actor, &host, write_path, "POST", Some(&body)); + let result = request(state.clone(), &host, write_path, "POST", Some(&auth), &body).await; + assert_eq!(result.0, StatusCode::OK, "{}", result.1); + assert_eq!( + result.1["outcomes"], + json!([{"status":"invalid"},{"status":"applied"},{"status":"applied"},{"status":"applied"}]) + ); + assert_eq!(result.1["channels"][0]["channel_id"], json!(channel)); + + let mut replies = Vec::new(); + for root in &roots { + let reply = EventBuilder::new(Kind::Custom(9), "reply") + .tags([Tag::parse(["e", &root.id.to_hex(), "", "root"]).unwrap()]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let at = |e: &nostr::Event| { + chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap() + }; + state + .db + .insert_event_with_thread_metadata( + community, + &reply, + Some(channel), + Some(buzz_db::event::ThreadMetadataParams { + event_id: reply.id.as_bytes(), + event_created_at: at(&reply), + 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: false, + }), + ) + .await + .unwrap(); + replies.push(reply); + } + let path = "/buzz/v1/me/threads?limit=1"; + let auth = proof(&actor, &host, path, "GET", None); + let result = request(state.clone(), &host, path, "GET", Some(&auth), b"").await; + assert_eq!(result.0, StatusCode::OK, "{}", result.1); + // Only the followed thread, unread since the follow. + assert_eq!( + result.1, + json!({"threads":[{"channel_id":channel,"root_id":roots[0].id.to_hex(), + "unread":true,"mentions":0,"read_through_id":null, + "latest_id":replies[0].id.to_hex()}],"next_cursor":null}) + ); + let path = "/buzz/v1/me/threads?cursor=not-a-cursor"; + let auth = proof(&actor, &host, path, "GET", None); + let result = request(state.clone(), &host, path, "GET", Some(&auth), b"").await; + assert_eq!(result.0, StatusCode::BAD_REQUEST, "{}", result.1); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn accessory_write_revocation_is_terminal_before_persistence() { @@ -495,7 +604,8 @@ async fn accessory_discovery_is_host_bound_and_opt_in() { doc["buzz_v1"], json!({"version":1,"base_path":"/buzz/v1","retention_seconds":1234, "max_channels":20,"max_intents":100, - "eligible_kinds":[9, 40002, 45001, 45003]}) + "eligible_kinds":[9, 40002, 45001, 45003], + "capabilities":["follow","threads"],"max_threads":50}) ); } } diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index d60771b2bc..2a885f14b2 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -97,6 +97,11 @@ pub struct BuzzV1Descriptor { /// Message kinds that count as unread and as latest activity. Clients /// classify live arrivals with this set instead of keeping a copy. pub eligible_kinds: [i32; 4], + /// Additions beyond the first v1 contract: `follow` (follow and unfollow + /// intents) and `threads` (`GET /me/threads`). Send one only when listed. + pub capabilities: Vec, + /// Maximum followed threads per threads page. + pub max_threads: usize, } /// Public capability descriptor for relay-proxied GIF search. @@ -382,7 +387,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st ); if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await { if state.config.buzz_v1_enabled { - use buzz_db::personal_read::{ELIGIBLE_KINDS, MAX_CHANNELS, MAX_INTENTS}; + use buzz_db::personal_read::{ELIGIBLE_KINDS, MAX_CHANNELS, MAX_INTENTS, MAX_THREADS}; info.buzz_v1 = Some(BuzzV1Descriptor { version: 1, base_path: crate::api::buzz_v1::BASE_PATH.to_owned(), @@ -390,6 +395,8 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st max_channels: MAX_CHANNELS, max_intents: MAX_INTENTS, eligible_kinds: ELIGIBLE_KINDS, + capabilities: vec!["follow".to_owned(), "threads".to_owned()], + max_threads: MAX_THREADS, }); } info.read_state_snapshot = Some(serde_json::json!({ diff --git a/docs/buzz-v1-read-state.md b/docs/buzz-v1-read-state.md index 5e0aab9a2c..58b988b1d1 100644 --- a/docs/buzz-v1-read-state.md +++ b/docs/buzz-v1-read-state.md @@ -12,9 +12,14 @@ application/nostr+json`, or `GET /info`) includes `buzz_v1` only when enabled: ```json {"buzz_v1":{"version":1,"base_path":"/buzz/v1","retention_seconds":2592000, -"max_channels":20,"max_intents":100,"eligible_kinds":[9,40002,45001,45003]}} +"max_channels":20,"max_intents":100,"eligible_kinds":[9,40002,45001,45003], +"capabilities":["follow","threads"],"max_threads":50}} ``` +`capabilities` lists additions to the first v1 contract: `follow` for the +`follow` and `unfollow` intents, `threads` for `GET /me/threads`. Use one only +when it is listed. + Use the requesting origin plus this relative prefix. Discovery is a configured capability, not a promise that the next request cannot fail. Unknown hosts and disabled deployments omit it, and a disabled deployment does not mount @@ -74,22 +79,23 @@ 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 that tag you with `p`, or every unread one in a DM (at most 100) | unread replies (at most 100) | +| `unread` | an unread timeline message counts | an unread reply counts | +| `mentions` | unread timeline messages that tag you with `p`, or every unread one in a DM (at most 100) | unread replies that tag you with `p`, or every unread one in a DM (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 reply to arrive | +| `latest_id` | the last eligible timeline 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 -row. `latest_id` is the anchor that reads its scope: "mark channel read" sends a -channel `mark_through` at the channel's `latest_id`, plus one thread -`mark_through` at each listed thread's `latest_id` to clear those too. The -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. +The timeline holds top-level messages and replies carrying `broadcast=1`; a +broadcast reply also counts on its thread row. No other reply makes the channel +row unread or becomes its `latest_id`. `latest_id` is the anchor that reads its +scope: "mark channel read" sends a channel `mark_through` at the channel's +`latest_id`, plus one thread `mark_through` at each listed thread's `latest_id` +to clear those too. The channel `latest_id` counts any author and ignores read +progress. Null means no timeline message was recorded since 0061, not that the +channel is empty. `read_through_id` may name a message that has since been +deleted. `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 +arrival first (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 @@ -107,7 +113,8 @@ 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. + otherwise the thread, which the author now follows. A broadcast reply reads + only its thread. - 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. @@ -115,11 +122,16 @@ in the same transaction: 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 +`broadcast=1` does not make anyone follow. Follows start when the relay that records them is deployed; there is no backfill. Desktop notification follows and mutes do not affect these counts. +The `follow` intent follows a thread explicitly, from now: replies that arrived +before it are read. Following a thread you already follow changes nothing. +`unfollow` stops counting it, and the stored message rules above no longer +follow you again, with two exceptions: posting in the thread follows you through +your post, and a mention follows you from just before it. + 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)). @@ -140,6 +152,27 @@ clock, relay arrival (see [Order](#order)). Three consequences: A later configuration expansion can change counts without having lost progress. +## Threads view + +`GET /buzz/v1/me/threads?limit=50&cursor=` lists the threads you follow +in joined, nondeleted channels, newest last reply first by arrival (then +`root_id`), read or not, at most 50 per page. Omit the cursor on the first +request and pass `next_cursor` back unchanged; null means the scan is done. A +thread with no reply yet is not listed. + +```json +{"threads":[{"channel_id":"","root_id":"<64-hex>","unread":false, + "mentions":0,"read_through_id":"<64-hex>","latest_id":"<64-hex>"}], + "next_cursor":""} +``` + +Each row has the sidebar thread row's fields and meanings plus its +`channel_id`, except that `unread` may be false and `mentions` is then 0. Pages +are separate snapshots: a thread that gets a new reply while you page moves +ahead of the cursor, so refresh from the first page after a change. Every +page probes each of your follows, so its cost grows with how many threads you +have ever followed, not with community size. + ## Fixed-operand writes `POST /buzz/v1/me/read-state` accepts 1–100 independent intents: @@ -147,7 +180,9 @@ A later configuration expansion can change counts without having lost progress. ```json {"intents":[ {"type":"mark_through","target":{"channel_id":""},"message_id":"<64-hex-event>"}, - {"type":"mark_through","target":{"channel_id":"","root_id":"<64-hex-root>"},"message_id":"<64-hex-event>"} + {"type":"mark_through","target":{"channel_id":"","root_id":"<64-hex-root>"},"message_id":"<64-hex-event>"}, + {"type":"follow","target":{"channel_id":"","root_id":"<64-hex-root>"}}, + {"type":"unfollow","target":{"channel_id":"","root_id":"<64-hex-root>"}} ]} ``` @@ -165,8 +200,9 @@ until its outcome is resolved. A mark-through validates a fixed message and advances its context's monotone frontier to that message's relay arrival (see [Order](#order)). A channel -anchor must be a top-level message; a thread anchor must be the thread's root -or one of its replies. The frontier records the anchor as `read_through_id` +anchor must be a top-level message or a broadcast reply; a thread anchor must +be the thread's root or one of its replies. `follow` and `unfollow` need a +thread target; a channel-only one is `invalid`. The frontier records the anchor as `read_through_id` when it advances; an anchor that arrived at or before the current frontier changes nothing. Opening a view is not itself a reading action; client dwell/focus policy determines when to send an actual observed anchor. Old or @@ -195,16 +231,16 @@ microsecond resolution. Three limits follow, none of which strands a badge: - Relay processes with different clocks can stamp out of true order by their skew. - Messages with the identical stamp are read together. +- A `follow` and the account's start use the database clock, so they can be + off from relay arrivals by the skew between the two. -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. +Counting walks forward from a position over indexes in arrival order: the +timeline's messages per channel and each thread's replies. A channel's +`latest_id` is exactly its last timeline arrival and a thread's its last reply +arrival, so marking through it reads everything counted. Mentions are indexed +by author time, which the relay accepts within 15 minutes of arrival, so a +mention walk starts 15 minutes of author time before the position: a mention +backdated further than that is unread but not counted as a mention. 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 @@ -225,28 +261,62 @@ device-local. a POST's rows for 100 channels fit the response limit. - 64 KiB write body; 1 MiB serialized API response. - 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 +- A caught-up reader costs one index probe per channel and per followed thread; a reader who is behind walks only what arrived after their position. +- 50 rows per threads page; each page probes every follow of the reader. - 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. -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`. 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. +Apply migrations 0056 through 0058 and 0061 (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`. 0061 adds `events.read_timeline` (set at insert +for timeline messages), `thread_metadata.received_at` (set at insert for +eligible replies) and `personal_read_frontiers.unfollowed`, plus three indexes: +`idx_events_read_timeline` for timeline walks, `idx_thread_metadata_read_arrival` +for thread walks and `idx_event_mentions_pubkey_channel_created` for mention +counts. There is no backfill: messages stored before the new relay serves, +including by older relay processes still running after 0061, are on neither +arrival index, so they never count as unread or become `latest_id`. 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. + +The column additions are catalog-only but briefly take an exclusive lock on +`events`, `thread_metadata` and `personal_read_frontiers`, bounded by the +migration's one-second `lock_timeout`; a busy writer fails the deployment, +which retries, rather than stalling ingestion. Building the indexes inside the +migration is bounded the same way, so on a database with data, add the columns +and prebuild the indexes before deploying, as for +[0049](thread-window-deployment.md). 0061 then skips both and only validates +the indexes. Run each statement on its own, retrying any that times out; +`CREATE INDEX CONCURRENTLY` cannot run in a transaction block: + +```sql +SET lock_timeout = '1s'; +ALTER TABLE events ADD COLUMN IF NOT EXISTS read_timeline BOOLEAN NOT NULL DEFAULT false; +ALTER TABLE thread_metadata ADD COLUMN IF NOT EXISTS received_at TIMESTAMPTZ; +ALTER TABLE personal_read_frontiers ADD COLUMN IF NOT EXISTS unfollowed BOOLEAN NOT NULL + DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea); +CREATE INDEX CONCURRENTLY idx_thread_metadata_read_arrival ON public.thread_metadata + (community_id, root_event_id, received_at DESC, event_id) WHERE received_at IS NOT NULL; +CREATE INDEX CONCURRENTLY idx_event_mentions_pubkey_channel_created ON public.event_mentions + (community_id, pubkey_hex, channel_id, event_created_at DESC); +-- events is partitioned: an invalid parent, then each partition concurrently. +CREATE INDEX idx_events_read_timeline ON ONLY public.events + (community_id, channel_id, received_at DESC, id) WHERE read_timeline AND deleted_at IS NULL; +-- For each partition p in +-- SELECT inhrelid::regclass FROM pg_inherits WHERE inhparent = 'public.events'::regclass: +CREATE INDEX CONCURRENTLY p_read_timeline ON p + (community_id, channel_id, received_at DESC, id) WHERE read_timeline AND deleted_at IS NULL; +ALTER INDEX idx_events_read_timeline ATTACH PARTITION p_read_timeline; +``` + +The parent becomes valid once every partition's index is attached. Verify the +three like 0049's, against the definitions the migration expects; recover a +failed build the same way. 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, diff --git a/migrations/0061_read_state_access_paths.sql b/migrations/0061_read_state_access_paths.sql new file mode 100644 index 0000000000..a1ec067769 --- /dev/null +++ b/migrations/0061_read_state_access_paths.sql @@ -0,0 +1,68 @@ +-- Read-state access paths: one index seek answers "newest countable message +-- here" and "what arrived after this position" for a channel timeline or a +-- thread, and "my mentions in this channel since then". Arrival order +-- (received_at), not author time. No backfill: rows written before this +-- migration are not on the new paths. +-- +-- events.read_timeline: an eligible message that counts on its channel +-- timeline (top-level, or a broadcast reply), set at insert. +-- thread_metadata.received_at: a reply's arrival, set at insert; NULL on root +-- stubs and earlier rows. +-- personal_read_frontiers.unfollowed: a sticky unfollow of a thread. +-- +-- Brownfield operators MUST add the three columns and prebuild the three +-- indexes concurrently as documented in docs/buzz-v1-read-state.md. Startup is +-- bounded like 0049: a large table fails deployment rather than blocking +-- ingestion. +SET LOCAL lock_timeout = '1s'; +SET LOCAL statement_timeout = '5s'; + +DO $$ +DECLARE + spec RECORD; +BEGIN + -- Even ADD COLUMN IF NOT EXISTS requests a writer-conflicting lock: skip + -- a column an operator added beforehand. + FOR spec IN SELECT * FROM (VALUES + ('events', 'read_timeline', 'BOOLEAN NOT NULL DEFAULT false'), + ('thread_metadata', 'received_at', 'TIMESTAMPTZ'), + ('personal_read_frontiers', 'unfollowed', + $d$BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea)$d$) + ) AS s(tbl, col, definition) LOOP + IF NOT EXISTS ( + SELECT 1 FROM pg_attribute + WHERE attrelid = ('public.' || spec.tbl)::regclass + AND attname = spec.col AND NOT attisdropped + ) THEN + EXECUTE format('ALTER TABLE public.%I ADD COLUMN %I %s', + spec.tbl, spec.col, spec.definition); + END IF; + END LOOP; + + -- A partitioned parent's definition reads ON ONLY; CREATE recurses to + -- every partition regardless. + FOR spec IN SELECT * FROM (VALUES + ('idx_events_read_timeline', + 'CREATE INDEX idx_events_read_timeline ON ONLY public.events USING btree (community_id, channel_id, received_at DESC, id) WHERE (read_timeline AND (deleted_at IS NULL))'), + ('idx_thread_metadata_read_arrival', + 'CREATE INDEX idx_thread_metadata_read_arrival ON public.thread_metadata USING btree (community_id, root_event_id, received_at DESC, event_id) WHERE (received_at IS NOT NULL)'), + ('idx_event_mentions_pubkey_channel_created', + 'CREATE INDEX idx_event_mentions_pubkey_channel_created ON public.event_mentions USING btree (community_id, pubkey_hex, channel_id, event_created_at DESC)') + ) AS s(name, definition) LOOP + -- IF NOT EXISTS still requests a writer-conflicting lock: skip CREATE + -- for a prebuilt index, then validate it. + IF to_regclass('public.' || spec.name) IS NULL THEN + EXECUTE replace(spec.definition, ' ON ONLY ', ' ON '); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname = spec.name + AND i.indisvalid AND i.indisready AND i.indislive + AND pg_get_indexdef(i.indexrelid) = spec.definition + ) THEN + RAISE EXCEPTION '% invalid or wrong definition; see docs/buzz-v1-read-state.md', spec.name; + END IF; + END LOOP; +END $$; diff --git a/schema/tables/public/event_mentions.sql b/schema/tables/public/event_mentions.sql index b0eb9dcb4a..df0ee76e53 100644 --- a/schema/tables/public/event_mentions.sql +++ b/schema/tables/public/event_mentions.sql @@ -18,5 +18,7 @@ CREATE INDEX idx_event_mentions_pubkey_created ON event_mentions (community_id, pubkey_hex, event_created_at DESC); CREATE INDEX idx_event_mentions_pubkey_kind_created ON event_mentions (community_id, pubkey_hex, event_kind, event_created_at DESC); +CREATE INDEX idx_event_mentions_pubkey_channel_created + ON event_mentions (community_id, pubkey_hex, channel_id, event_created_at DESC); CREATE INDEX idx_event_mentions_community_event ON event_mentions (community_id, event_id); diff --git a/schema/tables/public/events.sql b/schema/tables/public/events.sql index 10c1adc19e..936d7d737d 100644 --- a/schema/tables/public/events.sql +++ b/schema/tables/public/events.sql @@ -37,6 +37,8 @@ CREATE TABLE events ( d_tag TEXT, not_before BIGINT, delivered_at BIGINT, + -- An eligible message that counts on its channel's read-state timeline. + read_timeline BOOLEAN NOT NULL DEFAULT false, PRIMARY KEY (community_id, created_at, id) ) PARTITION BY RANGE (created_at); @@ -52,6 +54,10 @@ CREATE INDEX idx_events_community_pubkey_kind_created CREATE INDEX idx_events_community_kind_created ON events (community_id, kind, created_at DESC, id); CREATE INDEX idx_events_community_deleted ON events (community_id, deleted_at); +-- Read state: newest timeline message, and what arrived after a position. +CREATE INDEX idx_events_read_timeline + ON events (community_id, channel_id, received_at DESC, id) + WHERE read_timeline AND deleted_at IS NULL; -- Addressable (replaceable) and NIP-33 parameterized lookups. CREATE INDEX idx_events_addressable ON events (community_id, kind, pubkey, channel_id, deleted_at); diff --git a/schema/tables/public/personal_read_frontiers.sql b/schema/tables/public/personal_read_frontiers.sql index e62eb9c8b9..0aab16ad5c 100644 --- a/schema/tables/public/personal_read_frontiers.sql +++ b/schema/tables/public/personal_read_frontiers.sql @@ -6,6 +6,8 @@ CREATE TABLE personal_read_frontiers ( through_timestamp TIMESTAMPTZ NOT NULL, -- The message whose arrival is through_timestamp; NULL before 0058. through_message_id BYTEA CHECK (octet_length(through_message_id) = 32), + -- A sticky thread unfollow; thread rows only. + unfollowed BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea), PRIMARY KEY (community_id, actor, channel_id, root_id), FOREIGN KEY (community_id, actor) REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE, diff --git a/schema/tables/public/thread_metadata.sql b/schema/tables/public/thread_metadata.sql index 2c9ed9800a..94a162316e 100644 --- a/schema/tables/public/thread_metadata.sql +++ b/schema/tables/public/thread_metadata.sql @@ -15,6 +15,8 @@ CREATE TABLE thread_metadata ( descendant_count INT NOT NULL DEFAULT 0, last_reply_at TIMESTAMPTZ, broadcast BOOLEAN NOT NULL DEFAULT FALSE, + -- A reply's relay arrival; NULL on root stubs and rows before 0061. + received_at TIMESTAMPTZ, PRIMARY KEY (community_id, event_created_at, event_id), FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id) ); @@ -26,3 +28,6 @@ CREATE INDEX idx_thread_metadata_window CREATE INDEX idx_thread_metadata_channel_depth ON thread_metadata (community_id, channel_id, depth, event_created_at); CREATE INDEX idx_thread_metadata_event_id ON thread_metadata (community_id, event_id); +CREATE INDEX idx_thread_metadata_read_arrival + ON thread_metadata (community_id, root_event_id, received_at DESC, event_id) + WHERE received_at IS NOT NULL;