relay: re-follow on mention, O(1) thread rows, join floors threads

- A new mention re-follows an unfollowed thread, caught up to just before
  the mention; a mention while following leaves the position alone.
- Thread rows: `latest_id` is the last reply in display order and the sort
  key; `unread` is an existence probe. Nothing per thread grows with the
  unread count.
- Joining floors thread positions as well as the timeline, in both the
  sidebar and contexts, so a follow that survived leaving doesn't bring
  back replies from the absence.
- A top-level message stored with a depth-0 metadata row (workflow
  messages) marks its author read and follows mentioned members.
- Docs: re-follow rule, latest_id order, rolling-deploy note.

Signed-off-by: hMB2 <0400eb0da32afbbf16cdd1dee6ffd243587e8f3d6a2823fe7dfa95d16b002b7d@buzz.block.builderlab.xyz>

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

@@ -1880,7 +1880,14 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
             }
         }
 
-        if let (None, Some(channel)) = (&thread_meta, channel_id) {
+        // A top-level message, whether or not it arrives with a depth-0
+        // metadata row (workflow messages do); replies are recorded above.
+        let top_level = match &thread_meta {
+            None => channel_id,
+            Some(meta) if meta.parent_event_id.is_none() => Some(meta.channel_id),
+            Some(_) => None,
+        };
+        if let Some(channel) = top_level {
             crate::personal_read::record_message(
                 tx,
                 community_id,

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

@@ -279,7 +279,7 @@ async fn mark_thread_read_with_the_newest_displayed_reply_clears_a_late_reply()
         now - 40,
     )
     .await;
-    let late = reply(
+    let _late = reply(
         &db,
         &pool,
         community,
@@ -303,8 +303,8 @@ async fn mark_thread_read_with_the_newest_displayed_reply_clears_a_late_reply()
     assert_eq!((row.unread, row.threads.len()), (false, 1));
     assert_eq!(
         row.threads[0].latest_id,
-        late.id.to_hex(),
-        "the last arrival"
+        newest.id.to_hex(),
+        "the last in display order, not the last arrival"
     );
     assert_eq!(row.threads[0].mentions, 2);
 

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

@@ -51,12 +51,14 @@ impl Db {
             // account's start and absent before it: the channel's, never
             // before joining and absent for a non-member; and the thread's,
             // which exists only for threads the actor follows and includes any
-            // whole-channel cut.
+            // whole-channel cut and, for a member, joining: a follow that
+            // survived leaving does not bring back replies from the absence.
             let positions = sqlx::query(
                 "SELECT CASE WHEN m.joined_at IS NOT NULL AND s.started IS NOT NULL
                         THEN GREATEST(cf.through_timestamp, m.joined_at, s.started) END AS channel,
                     CASE WHEN tf.root_id IS NOT NULL AND s.started IS NOT NULL
-                        THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp, s.started)
+                        THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp,
+                            m.joined_at, s.started)
                         END AS thread
                  FROM (SELECT (SELECT started_at FROM personal_read_accounts
                     WHERE community_id=$1 AND actor=$2) AS started) s

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

@@ -28,8 +28,11 @@ pub(crate) enum Place<'a> {
 /// - a reply also makes the root's author, and in a DM (at most nine
 ///   members) every member, follow the thread.
 ///
-/// Nobody but the author is re-followed: an unfollowed row stays unfollowed,
-/// even for a new mention (whether a mention should re-follow is open).
+/// A new mention re-follows an unfollowed thread, catching up to just before
+/// the mention: chatter from while unfollowed stays muted, but the mention
+/// shows. A mentioned member already following keeps their position. Anyone
+/// else the reply makes follow keeps an existing row as it is, so an
+/// unfollow sticks for ordinary replies.
 ///
 /// Everyone but the author starts just before the message, so it is unread.
 pub(crate) async fn record_message(
@@ -53,8 +56,8 @@ pub(crate) async fn record_message(
         Place::Reply { root, broadcast } => (root, true, broadcast),
     };
     // Rank 0 is the replying author, who follows (again) and has read; 1 a
-    // mentioned member and 2 anyone else the reply makes follow, whose
-    // existing rows are left alone. Every insert's fence and foreign-key
+    // mentioned member, who follows again if they had unfollowed; 2 anyone
+    // else the reply makes follow, whose existing rows are left alone. Every insert's fence and foreign-key
     // checks run at the end of the statement, after the account rows exist.
     sqlx::query(
         "WITH joined AS (
@@ -92,10 +95,18 @@ pub(crate) async fn record_message(
             ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
             SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
                 excluded.through_timestamp), following=true
+         ), mentioned AS (
+            INSERT INTO personal_read_frontiers
+                (community_id, actor, channel_id, root_id, through_timestamp)
+            SELECT $1, actor, $2, $3, through FROM members WHERE rank=1
+            ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
+            SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
+                excluded.through_timestamp), following=true
+            WHERE NOT personal_read_frontiers.following
          )
          INSERT INTO personal_read_frontiers
             (community_id, actor, channel_id, root_id, through_timestamp)
-         SELECT $1, actor, $2, $3, through FROM members WHERE rank>0
+         SELECT $1, actor, $2, $3, through FROM members WHERE rank=2
          ON CONFLICT DO NOTHING",
     )
     .bind(community.as_uuid())

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

@@ -107,8 +107,8 @@ pub struct ThreadReadSummary {
     /// The last reply, in display order, at or before the actor's thread
     /// position. None when the actor has read no reply.
     pub read_through_id: Option<String>,
-    /// Last unread reply to arrive: marking through it reads the thread, and
-    /// clients fetch it by ID for a preview.
+    /// The last reply in display order: marking through it reads the thread,
+    /// and clients fetch it by ID for a preview.
     pub latest_id: String,
 }
 

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

@@ -353,10 +353,34 @@ async fn rejoining_starts_caught_up_again() {
     .execute(&pool)
     .await
     .unwrap();
+    // The member follows a thread; the follow survives leaving.
+    let root = post(&db, community, channel, &member, now(), vec![]).await;
+    reply(
+        &db,
+        community,
+        channel,
+        &root,
+        &member,
+        now(),
+        vec![],
+        false,
+    )
+    .await;
     db.remove_member(community, channel, &pk, &pk)
         .await
         .unwrap();
     let away = post(&db, community, channel, &Keys::generate(), now(), vec![]).await;
+    let away_reply = reply(
+        &db,
+        community,
+        channel,
+        &root,
+        &Keys::generate(),
+        now(),
+        mention(&member),
+        false,
+    )
+    .await;
     // Re-adding an active member (a role change) must not move the position.
     let owner_pk = owner.public_key().to_bytes();
     db.add_member(
@@ -374,10 +398,23 @@ async fn rejoining_starts_caught_up_again() {
         .unwrap();
     let row = sidebar(&db, community, &member).await;
     assert!(!row.unread, "messages from while away are read");
+    assert!(row.threads.is_empty(), "so are replies in followed threads");
     assert_eq!(
         status(&db, community, &member, channel, None, &[&away]).await,
         ["read"]
     );
+    assert_eq!(
+        status(
+            &db,
+            community,
+            &member,
+            channel,
+            Some(&root),
+            &[&away_reply]
+        )
+        .await,
+        ["read"]
+    );
     post(&db, community, channel, &Keys::generate(), now(), vec![]).await;
     assert!(sidebar(&db, community, &member).await.unread);
 }
@@ -602,6 +639,75 @@ async fn posting_marks_read_and_read_through_skips_deleted_and_auxiliary() {
     assert!(!empty.unread && empty.mentions == 0 && empty.read_through_id.is_none());
 }
 
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn top_level_message_with_depth_zero_metadata_marks_read_and_follows() {
+    // Workflow messages store a depth-0 metadata row with no parent.
+    let (db, _, community, channel, actor, _) = fixture().await;
+    let base = now();
+    let author = Keys::generate();
+    db.add_member(
+        community,
+        channel,
+        &author.public_key().to_bytes(),
+        MemberRole::Member,
+        None,
+    )
+    .await
+    .unwrap();
+    let insert = |who: Keys, at: u64, tags: Vec<Tag>| {
+        let db = db.clone();
+        async move {
+            let event = EventBuilder::new(Kind::Custom(9), format!("workflow at {at}"))
+                .tags(tags)
+                .custom_created_at(nostr::Timestamp::from(at))
+                .sign_with_keys(&who)
+                .unwrap();
+            db.insert_event_with_thread_metadata(
+                community,
+                &event,
+                Some(channel),
+                Some(crate::event::ThreadMetadataParams {
+                    event_id: event.id.as_bytes(),
+                    event_created_at: chrono::DateTime::from_timestamp(at as i64, 0).unwrap(),
+                    channel_id: channel,
+                    parent_event_id: None,
+                    parent_event_created_at: None,
+                    root_event_id: None,
+                    root_event_created_at: None,
+                    depth: 0,
+                    broadcast: false,
+                }),
+            )
+            .await
+            .unwrap();
+            event
+        }
+    };
+    let own = insert(actor.clone(), base + 1, vec![]).await;
+    let row = sidebar(&db, community, &actor).await;
+    assert!(!row.unread, "posting read through it");
+    assert_eq!(row.read_through_id, Some(own.id.to_hex()));
+
+    let root = insert(author.clone(), base + 2, mention(&actor)).await;
+    let row = sidebar(&db, community, &actor).await;
+    assert_eq!((row.unread, row.mentions), (true, 1));
+    reply(
+        &db,
+        community,
+        channel,
+        &root,
+        &author,
+        base + 3,
+        vec![],
+        false,
+    )
+    .await;
+    let row = sidebar(&db, community, &actor).await;
+    assert_eq!(row.threads.len(), 1, "the mention followed its thread");
+    assert_eq!(row.threads[0].root_id, root.id.to_hex());
+}
+
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn position_is_monotonic_and_rejects_malformed_anchors() {

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

@@ -80,7 +80,9 @@ impl Db {
         // the actor's first intent, so nothing counts). A channel's is its
         // frontier, never earlier than joining: a new member starts caught
         // up. A thread's is its row, which exists only for threads the actor
-        // follows (see `membership`), and a whole-channel cut covers them all.
+        // follows (see `membership`), and a whole-channel cut and joining
+        // cover them all, so a follow that survived leaving does not bring
+        // back replies from the absence.
         //
         // - unread: the first timeline message past the position
         //   (idx_events_community_channel_created);
@@ -90,13 +92,16 @@ impl Db {
         //   arrived at or before the position. Anything that arrived by then
         //   was authored no later than the position plus the skew;
         // - threads: followed roots whose last reply arrived after the
-        //   thread position, then their unread replies and mentions.
+        //   thread position and that have an unread reply (an existence
+        //   probe), then their mentions; `latest_id` is the last reply in
+        //   display order (idx_thread_metadata_window), so no part of a
+        //   thread row grows with its unread count.
         let sql = format!(
             r#"WITH roster AS MATERIALIZED (
                 SELECT c.id, c.name, c.channel_type::text AS channel_type,
                     c.archived_at IS NOT NULL AS archived, cm.hidden_at IS NOT NULL AS hidden,
                     GREATEST(cf.through_timestamp, cm.joined_at, s.started) AS position,
-                    GREATEST(cf.threads_through_timestamp, s.started) AS threads_floor
+                    GREATEST(cf.threads_through_timestamp, cm.joined_at, s.started) AS threads_floor
                 FROM channel_members cm JOIN channels c
                     ON c.community_id=cm.community_id AND c.id=cm.channel_id
                 CROSS JOIN (SELECT COALESCE((SELECT started_at FROM personal_read_accounts
@@ -133,15 +138,13 @@ impl Db {
                     AND root.depth=0 AND root.event_id=tf.root_id
                 CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, r.threads_floor) AS position) p
                 CROSS JOIN LATERAL (
-                    SELECT (array_agg(encode(e.id,'hex') ORDER BY e.received_at DESC,e.id))[1] AS id,
-                        max(e.created_at) AS at
+                    SELECT encode(tm.event_id,'hex') AS id, tm.event_created_at AS at
                     FROM thread_metadata tm JOIN events e ON e.community_id=$1
                         AND e.created_at=tm.event_created_at AND e.id=tm.event_id
                     WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id
                         AND tm.channel_id=r.id AND tm.event_id<>tf.root_id
-                        AND tm.event_created_at >= p.position-$7
-                        AND e.received_at > p.position AND e.kind=ANY($6)
-                        AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast)
+                        AND e.kind=ANY($6) AND e.deleted_at IS NULL
+                    ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1
                 ) n
                 CROSS JOIN LATERAL (
                     SELECT count(*) AS mentions
@@ -163,7 +166,15 @@ impl Db {
                 ) rt ON true
                 WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
                     AND tf.root_id<>''::bytea AND tf.following
-                    AND root.last_reply_received_at > p.position AND n.id IS NOT NULL
+                    AND root.last_reply_received_at > p.position
+                    AND EXISTS (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 tm.community_id=$1 AND tm.root_event_id=tf.root_id
+                            AND tm.channel_id=r.id AND tm.event_id<>tf.root_id
+                            AND tm.event_created_at >= p.position-$7
+                            AND e.received_at > p.position AND e.kind=ANY($6)
+                            AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast))
              ) threads ON true
              ORDER BY r.id"#
         );

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

@@ -466,8 +466,10 @@ async fn thread_summaries_order_newest_first_and_list_every_unread_thread() {
             roots[5].id.to_hex()
         ]
     );
-    // Ordered by its newest reply's author time, anchored at its last arrival.
-    assert_eq!(row.threads[2].latest_id, last.unwrap().id.to_hex());
+    // Ordered by, and anchored at, its last reply in display order, not the
+    // last to arrive.
+    assert_eq!(row.threads[2].latest_id, anchors[2].id.to_hex());
+    assert_ne!(row.threads[2].latest_id, last.unwrap().id.to_hex());
 
     // Reading a listed thread through its anchor removes only it.
     let first = &row.threads[0];
@@ -651,33 +653,53 @@ async fn follow_starts_caught_up_and_unfollow_is_sticky() {
     assert_eq!(row.threads.len(), 1);
     assert_eq!(row.threads[0].latest_id, next.id.to_hex());
 
-    // Unfollow keeps the row; later replies, mentions included, don't
-    // re-follow (whether a new mention should is an open product question).
+    // Unfollow keeps the row; later replies don't re-follow.
     assert_eq!(
         apply(&db, community, &actor, follow(channel, &root, false)).await,
         IntentOutcome::Applied
     );
     assert!(sidebar(&db, community, &actor).await.threads.is_empty());
-    say(base + 2, vec![]).await;
-    say(base + 3, mention(&actor)).await;
+    let chatter = say(base + 2, vec![]).await;
     assert_eq!(following(&pool, community, &actor).await, [false]);
     assert!(sidebar(&db, community, &actor).await.threads.is_empty());
 
-    // The actor's own reply re-follows, read through it.
+    // A new mention re-follows, catching up to just before it: the chatter
+    // from while unfollowed stays read, the mention shows.
+    let mentioned = say(base + 3, mention(&actor)).await;
+    assert_eq!(following(&pool, community, &actor).await, [true]);
+    let row = sidebar(&db, community, &actor).await;
+    assert_eq!(row.threads.len(), 1);
+    assert_eq!(
+        (
+            row.threads[0].mentions,
+            row.threads[0].latest_id.clone(),
+            row.threads[0].read_through_id.clone()
+        ),
+        (1, mentioned.id.to_hex(), Some(chatter.id.to_hex()))
+    );
+    // A mention while following leaves the position alone.
+    let again = say(base + 4, mention(&actor)).await;
+    let row = sidebar(&db, community, &actor).await;
+    assert_eq!(
+        (row.threads[0].mentions, row.threads[0].latest_id.clone()),
+        (2, again.id.to_hex())
+    );
+
+    // The actor's own reply reads the thread through it.
     let own = reply(
         &db,
         community,
         channel,
         &root,
         &actor,
-        base + 4,
+        base + 5,
         vec![],
         false,
     )
     .await;
     assert_eq!(following(&pool, community, &actor).await, [true]);
     assert!(sidebar(&db, community, &actor).await.threads.is_empty());
-    say(base + 5, vec![]).await;
+    say(base + 6, vec![]).await;
     let row = sidebar(&db, community, &actor).await;
     assert_eq!(row.threads.len(), 1);
     assert_eq!(
@@ -686,7 +708,7 @@ async fn follow_starts_caught_up_and_unfollow_is_sticky() {
             row.threads[0].read_through_id.clone()
         ),
         (0, Some(own.id.to_hex())),
-        "the earlier mention was read when the actor replied"
+        "the earlier mentions were read when the actor replied"
     );
 
     // Following again catches up; a target without a root is invalid.

docs/buzz-v1-read-state.md

@@ -79,11 +79,12 @@ thread with an unread reply:
 - `read_through_id`: the last message in that scope, in display order, at or
   before your read position, or null when there is none. Draw the "new"
   divider below it and open the view there.
-- `latest_id` (threads only): the last unread reply to arrive. Fetch it by ID
-  for a preview; marking the thread through it reads every listed reply.
+- `latest_id` (threads only): the last reply in display order. Fetch it by ID
+  for a preview; marking the thread through it reads every reply above it,
+  late arrivals included.
 
-`threads` lists every followed thread with an unread reply, newest reply
-(author time) first, then by `root_id`. There is no cap. A badge or a threads
+`threads` lists every followed thread with an unread reply, by `latest_id`
+(newest author time first), then by `root_id`. There is no cap. A badge or a threads
 page is computed on the client from these rows. Clients receive message IDs
 only; positions and timestamps stay relay-internal. No message bytes are
 included.
@@ -107,14 +108,15 @@ The relay updates read state when it stores an eligible message:
   any member replies, or with a `follow` intent. A new follow starts just
   before the message that caused it, so that message is unread.
 - **Unfollow is sticky.** `unfollow` keeps the row with `following=false`, so
-  later replies and mentions do not follow you again. Your own reply, or
-  `follow`, does. Whether a new mention should re-follow is an open product
-  question; today it does not.
+  later replies do not follow you again. Your own reply, `follow`, or a new
+  mention does. A mention catches the thread up to just before itself: the
+  replies from while you were unfollowed stay read, and the mention shows.
 
 An account starts caught up at its first read intent (`started_at`): every
 position is floored there, so history from before then never counts, and
 nothing counts before the first intent. Joining a channel starts that channel
-caught up.
+caught up, its followed threads included: a follow that survived leaving does
+not bring back replies from your absence.
 
 ## Explicit contexts
 
@@ -231,6 +233,11 @@ private tables, adds `thread_metadata.last_reply_received_at` and
 position can reach), and one index on `event_mentions`. Ingest writes the
 author's own position and follow rows in the same transaction as the message.
 
+During a rolling deploy, relay processes still on the old code write
+`event_mentions` rows without `received_at`/`root_id` and do not mark posting
+read. Those mentions never count, and an author's own message from such a
+process can show unread to them until they next read the scope.
+
 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,
 actor, channel or position values should become metric labels.