feat(read-state): count thread replies only in followed threads

crates/buzz-db/src/store/event.rs

@@ -447,6 +447,16 @@ pub async fn insert_event_in_transaction(
 ) -> Result<(StoredEvent, bool)> {
     let result = insert_event_on(tx, event, channel_id).await?;
     if result.1 {
+        if let Some(channel) = channel_id {
+            crate::personal_read::record_message(
+                tx,
+                event,
+                channel,
+                crate::personal_read::Place::TopLevel,
+                result.0.received_at,
+            )
+            .await?;
+        }
         crate::operator_listener::enqueue_mentions_in_transaction(tx, event).await?;
     }
     Ok(result)
@@ -1872,6 +1882,22 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
             }
         }
 
+        let place = match &thread_meta {
+            Some(meta) => Some((
+                meta.channel_id,
+                match meta.parent_event_id {
+                    Some(parent) => crate::personal_read::Place::Reply {
+                        root: meta.root_event_id.unwrap_or(parent),
+                    },
+                    // Workflow messages store a depth-0 metadata row.
+                    None => crate::personal_read::Place::TopLevel,
+                },
+            )),
+            None => channel_id.map(|channel| (channel, crate::personal_read::Place::TopLevel)),
+        };
+        if let Some((channel, place)) = place {
+            crate::personal_read::record_message(tx, event, channel, place, received_at).await?;
+        }
         crate::operator_listener::enqueue_mentions_in_transaction(tx, event).await?;
     }
 

crates/buzz-db/src/store/personal_read/arrival_postgres_tests.rs

@@ -1,6 +1,9 @@
 //! Read progress follows relay arrival (`received_at`), never author time.
 //! Each case sets every arrival explicitly: back-to-back inserts share a clock.
-use super::{postgres_tests::fixture, *};
+use super::{
+    postgres_tests::{fixture, insert_reply},
+    *,
+};
 use crate::Db;
 use buzz_core::CommunityId;
 use nostr::{EventBuilder, Keys, Kind, Tag};
@@ -45,7 +48,8 @@ async fn post(
     event
 }
 
-/// A reply to `root` that mentions the actor, so it counts without membership.
+/// A reply to `root` that mentions the actor, so the actor follows the thread
+/// from just before the earliest such reply to arrive.
 #[allow(clippy::too_many_arguments)]
 async fn reply(
     db: &Db,
@@ -57,12 +61,29 @@ async fn reply(
     authored: u64,
     arrived: u64,
 ) -> nostr::Event {
-    let event = post(db, pool, community, channel, actor, authored, arrived).await;
-    sqlx::query("INSERT INTO thread_metadata (community_id,event_id,event_created_at,channel_id,root_event_id,parent_event_id,depth)
-        VALUES ($1,$2,to_timestamp($3),$4,$5,$5,1)")
-        .bind(community.as_uuid()).bind(event.id.as_bytes().as_slice())
-        .bind(event.created_at.as_secs() as f64)
-        .bind(channel).bind(root.id.as_bytes().as_slice()).execute(pool).await.unwrap();
+    let event = EventBuilder::new(Kind::Custom(9), format!("authored {authored}"))
+        .tags([Tag::public_key(actor.public_key())])
+        .custom_created_at(nostr::Timestamp::from(authored))
+        .sign_with_keys(&Keys::generate())
+        .unwrap();
+    insert_reply(db, community, channel, root, &event).await;
+    arrive(pool, community, &event, arrived).await;
+    // Ingest started the follow just before the real arrival; move it to
+    // just before the rewritten one.
+    let moved = sqlx::query(
+        "UPDATE personal_read_frontiers
+         SET through_timestamp=LEAST(through_timestamp, to_timestamp($4)-interval '1 microsecond')
+         WHERE community_id=$1 AND actor=$2 AND root_id=$3",
+    )
+    .bind(community.as_uuid())
+    .bind(actor.public_key().to_bytes().as_slice())
+    .bind(root.id.as_bytes().as_slice())
+    .bind(arrived as f64)
+    .execute(pool)
+    .await
+    .unwrap()
+    .rows_affected();
+    assert_eq!(moved, 1, "the mention follows the thread");
     event
 }
 
@@ -233,11 +254,39 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
     assert!(row.threads.is_empty());
 }
 
-/// The shallow latest probe reads the newest `MAX_CHANNEL_SCAN + 1` events by
-/// author time, of any kind. A late arrival with an older author time than
-/// that many events is counted unread, so the unread scan must supply the
-/// anchor or Mark as read with the sidebar anchor leaves it.
-async fn mark_as_read_behind_fillers(fillers: u64) -> (bool, u32) {
+/// The latest probe reads the newest `MAX_TIMELINE_SCAN` events by author
+/// time, of any kind. A late arrival with an older author time than the
+/// uncounted events after it must still be the sidebar anchor, or Mark as read
+/// with that anchor leaves it unread.
+async fn reactions(
+    pool: &PgPool,
+    db: &Db,
+    community: CommunityId,
+    channel: Uuid,
+    count: u64,
+    now: u64,
+) {
+    for i in 0..count {
+        // Reactions are not counted, but the probe's LIMIT sees them.
+        let filler = EventBuilder::new(Kind::Custom(7), "+")
+            .custom_created_at(nostr::Timestamp::from(now - 500 + i % 400))
+            .sign_with_keys(&Keys::generate())
+            .unwrap();
+        db.insert_event(community, &filler, Some(channel))
+            .await
+            .unwrap();
+    }
+    sqlx::query("UPDATE events SET received_at=to_timestamp($2) WHERE community_id=$1 AND kind=7")
+        .bind(community.as_uuid())
+        .bind((now - 50) as f64)
+        .execute(pool)
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn mark_as_read_clears_a_late_arrival_behind_reactions() {
     let (db, pool, community, channel, actor, first) = fixture().await;
     let now = first.created_at.as_secs();
     arrive(&pool, community, &first, now - 60).await;
@@ -248,17 +297,8 @@ async fn mark_as_read_behind_fillers(fillers: u64) -> (bool, u32) {
         mark_through(channel, None, &first.id.to_hex()),
     )
     .await;
-    for i in 0..fillers {
-        // Reactions are not counted, but the probe's LIMIT sees them.
-        let filler = EventBuilder::new(Kind::Custom(7), "+")
-            .custom_created_at(nostr::Timestamp::from(now - 500 + i))
-            .sign_with_keys(&Keys::generate())
-            .unwrap();
-        db.insert_event(community, &filler, Some(channel))
-            .await
-            .unwrap();
-        arrive(&pool, community, &filler, now - 50).await;
-    }
+    // More than main's old 256-event probe.
+    reactions(&pool, &db, community, channel, 300, now).await;
     post(&db, &pool, community, channel, &actor, now - 600, now - 10).await;
     assert_eq!(counts(&db, community, &actor).await, (true, 1));
 
@@ -267,29 +307,14 @@ async fn mark_as_read_behind_fillers(fillers: u64) -> (bool, u32) {
         .latest_id
         .expect("a channel with messages has a latest message");
     apply(&db, community, &actor, mark_through(channel, None, &anchor)).await;
-    counts(&db, community, &actor).await
-}
-
-/// Control: with one slot to spare the late arrival is inside the probe.
-#[tokio::test]
-#[ignore = "requires Postgres"]
-async fn mark_as_read_clears_a_late_arrival_inside_the_latest_probe() {
-    let fillers = MAX_CHANNEL_SCAN as u64 - 1;
-    assert_eq!(mark_as_read_behind_fillers(fillers).await, (false, 0));
-}
-
-#[tokio::test]
-#[ignore = "requires Postgres"]
-async fn mark_as_read_clears_a_late_arrival_outside_the_latest_probe() {
-    let fillers = MAX_CHANNEL_SCAN as u64;
-    assert_eq!(mark_as_read_behind_fillers(fillers).await, (false, 0));
+    assert_eq!(counts(&db, community, &actor).await, (false, 0));
 }
 
-/// When the newest 257 events are all uncounted, the only message is still in
-/// the unread scan: the summary reports it as latest and unread.
+/// Past the budget the probe finds no anchor, and nothing counts: a badge
+/// with no latest to mark could never be cleared.
 #[tokio::test]
 #[ignore = "requires Postgres"]
-async fn latest_message_behind_a_full_probe_of_reactions_is_found() {
+async fn a_message_behind_a_full_budget_of_reactions_is_not_a_badge_without_an_anchor() {
     let (db, pool, community, channel, actor, first) = fixture().await;
     let now = first.created_at.as_secs();
     let demoted = sqlx::query("UPDATE events SET kind=7 WHERE community_id=$1 AND id=$2")
@@ -300,21 +325,20 @@ async fn latest_message_behind_a_full_probe_of_reactions_is_found() {
         .unwrap()
         .rows_affected();
     assert_eq!(demoted, 1);
-    let only = post(&db, &pool, community, channel, &actor, now - 600, now - 10).await;
-    for i in 0..MAX_CHANNEL_SCAN as u64 {
-        let filler = EventBuilder::new(Kind::Custom(7), "+")
-            .custom_created_at(nostr::Timestamp::from(now - 500 + i))
-            .sign_with_keys(&Keys::generate())
-            .unwrap();
-        db.insert_event(community, &filler, Some(channel))
-            .await
-            .unwrap();
-        arrive(&pool, community, &filler, now - 50).await;
-    }
+    post(&db, &pool, community, channel, &actor, now - 600, now - 10).await;
+    reactions(
+        &pool,
+        &db,
+        community,
+        channel,
+        MAX_TIMELINE_SCAN as u64,
+        now,
+    )
+    .await;
 
     let row = sidebar(&db, community, &actor).await;
-    assert_eq!(row.latest_id, Some(only.id.to_hex()));
-    assert_eq!((row.unread, row.mentions), (true, 1));
+    assert_eq!(row.latest_id, None);
+    assert_eq!((row.unread, row.mentions), (false, 0));
 }
 
 /// Store `event` as having arrived at exactly `seconds` plus `micros`. Built

crates/buzz-db/src/store/personal_read/follows.rs

@@ -0,0 +1,113 @@
+//! Read state that ingest writes. A thread frontier row exists only for a
+//! thread its actor follows; only this module creates one.
+use super::model::ELIGIBLE_KINDS;
+use crate::{AdmittedTx, Result};
+use chrono::{DateTime, Utc};
+use nostr::Event;
+use uuid::Uuid;
+
+/// Where a newly stored channel message sits.
+#[derive(Clone, Copy)]
+pub(crate) enum Place<'a> {
+    /// A top-level timeline message.
+    TopLevel,
+    /// A reply in the thread rooted at `root`.
+    Reply { root: &'a [u8] },
+}
+
+/// Record what a newly stored eligible message changes, in its transaction:
+/// - its author has read through it: the timeline for a top-level message,
+///   otherwise the thread, which the author then follows;
+/// - each channel member it mentions follows its thread (for a top-level
+///   message, the thread rooted at it);
+/// - a reply makes the root's author, and in a DM every member, follow too.
+///
+/// Everyone but the author starts just before the message, so it is unread
+/// to them. A follower keeps an existing position.
+pub(crate) async fn record_message(
+    tx: &mut AdmittedTx,
+    event: &Event,
+    channel: Uuid,
+    place: Place<'_>,
+    received_at: DateTime<Utc>,
+) -> Result<()> {
+    if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
+        return Ok(());
+    }
+    let community = tx.community();
+    let id = event.id.as_bytes().as_slice();
+    let mentioned: Vec<Vec<u8>> = event
+        .tags
+        .public_keys()
+        .map(|key| key.to_bytes().to_vec())
+        .collect();
+    let (read, thread, reply) = match place {
+        Place::TopLevel => (&[][..], id, false),
+        Place::Reply { root } => (root, root, true),
+    };
+    // Accounts, then frontiers, each in key order, so concurrent messages that
+    // involve the same people cannot deadlock: the frontier insert joins the
+    // finished account insert. Foreign-key checks run at the end of the
+    // statement, after the account rows exist.
+    sqlx::query(
+        "WITH involved (actor, root_id, through, message) AS (
+            SELECT $4::bytea, $5::bytea, $6::timestamptz, $3::bytea
+            UNION ALL
+            SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL FROM channel_members cm
+            WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
+                AND cm.pubkey=ANY($8) AND cm.pubkey<>$4
+            UNION ALL
+            SELECT e.pubkey, $7, $6-interval '1 microsecond', NULL FROM events e
+            WHERE $9 AND e.community_id=$1 AND e.channel_id=$2 AND e.id=$7 AND e.pubkey<>$4
+            UNION ALL
+            SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL
+            FROM channel_members cm JOIN channels c
+                ON c.community_id=cm.community_id AND c.id=cm.channel_id
+            WHERE $9 AND cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
+                AND c.channel_type='dm' AND cm.pubkey<>$4
+         ), rows AS (
+            -- An existing follower keeps their position, so is not touched.
+            -- Existence probes are LIMIT 1 laterals: always one index lookup.
+            SELECT DISTINCT ON (i.actor, i.root_id) i.* FROM involved i
+            LEFT JOIN LATERAL (
+                SELECT true AS found FROM personal_read_frontiers f WHERE f.community_id=$1
+                    AND f.actor=i.actor AND f.channel_id=$2 AND f.root_id=i.root_id LIMIT 1
+            ) f ON true
+            WHERE i.message IS NOT NULL OR f.found IS NULL
+            ORDER BY i.actor, i.root_id
+         ), accounts AS (
+            INSERT INTO personal_read_accounts (community_id, actor)
+            SELECT $1, r.actor FROM (SELECT DISTINCT actor FROM rows) r
+            LEFT JOIN LATERAL (
+                SELECT true AS found FROM personal_read_accounts a
+                WHERE a.community_id=$1 AND a.actor=r.actor LIMIT 1
+            ) a ON true
+            WHERE a.found IS NULL ORDER BY r.actor
+            ON CONFLICT DO NOTHING RETURNING 1
+         )
+         INSERT INTO personal_read_frontiers
+            (community_id, actor, channel_id, root_id, through_timestamp, through_message_id)
+         SELECT $1, actor, $2, root_id, through, message
+         FROM rows CROSS JOIN (SELECT count(*) FROM accounts) written
+         ORDER BY actor, root_id
+         ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
+         SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
+                excluded.through_timestamp),
+            through_message_id=CASE
+                WHEN excluded.through_timestamp > personal_read_frontiers.through_timestamp
+                THEN excluded.through_message_id ELSE personal_read_frontiers.through_message_id END
+         WHERE excluded.through_message_id IS NOT NULL",
+    )
+    .bind(community.as_uuid())
+    .bind(channel)
+    .bind(id)
+    .bind(event.pubkey.to_bytes().as_slice())
+    .bind(read)
+    .bind(received_at)
+    .bind(thread)
+    .bind(&mentioned)
+    .bind(reply)
+    .execute(tx.conn())
+    .await?;
+    Ok(())
+}