relay: channels answer "anything new?" from their last timeline arrival

channels.last_timeline_received_at mirrors thread_metadata's
last_reply_received_at: ingest raises it (GREATEST) for eligible top-level
messages and broadcast replies as the last statement of the transaction,
and the sidebar's channel unread probe runs only when it is past the
position. A caught-up reader in a mostly-replies channel no longer walks
every newer reply. No backfill: started_at floors every position after
the migration, so NULL correctly reads as nothing new.

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

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

@@ -456,6 +456,16 @@ pub async fn insert_event_in_transaction(
             .await?;
         }
         crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
+        if let Some(channel) = channel_id {
+            crate::personal_read::record_timeline_arrival(
+                tx,
+                community_id,
+                event,
+                channel,
+                result.0.received_at,
+            )
+            .await?;
+        }
     }
     Ok(result)
 }
@@ -1899,6 +1909,24 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
             .await?;
         }
         crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
+        // Last, so the channel row lock is held only until commit.
+        let timeline = match &thread_meta {
+            None => channel_id,
+            Some(meta) if meta.parent_event_id.is_none() || (meta.depth == 1 && meta.broadcast) => {
+                Some(meta.channel_id)
+            }
+            Some(_) => None,
+        };
+        if let Some(channel) = timeline {
+            crate::personal_read::record_timeline_arrival(
+                tx,
+                community_id,
+                event,
+                channel,
+                received_at,
+            )
+            .await?;
+        }
     }
 
     Ok((

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

@@ -121,3 +121,31 @@ pub(crate) async fn record_message(
     .await?;
     Ok(())
 }
+
+/// Note that an eligible timeline message (top-level, or a broadcast reply)
+/// arrived in `channel`. Run it as the ingest transaction's last statement:
+/// concurrent posts to one channel queue on this row lock until commit.
+/// `GREATEST` makes their order irrelevant. Key-share locks from inserts that
+/// reference the channel do not conflict with this non-key update.
+pub(crate) async fn record_timeline_arrival(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    event: &Event,
+    channel: Uuid,
+    received_at: DateTime<Utc>,
+) -> Result<()> {
+    if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
+        return Ok(());
+    }
+    sqlx::query(
+        "UPDATE channels SET last_timeline_received_at =
+            GREATEST(last_timeline_received_at, $3)
+         WHERE community_id=$1 AND id=$2",
+    )
+    .bind(community.as_uuid())
+    .bind(channel)
+    .bind(received_at)
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}

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

@@ -12,7 +12,7 @@ mod model;
 mod projection;
 mod writes;
 
-pub(crate) use membership::{record_message, Place};
+pub(crate) use membership::{record_message, record_timeline_arrival, Place};
 pub use model::*;
 
 #[cfg(test)]

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

@@ -639,6 +639,74 @@ 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 channel_unread_follows_timeline_arrivals_only() {
+    let (db, pool, community, channel, actor, root) = fixture().await;
+    let base = now();
+    let other = Keys::generate();
+    let last = |pool: PgPool| async move {
+        sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
+            "SELECT last_timeline_received_at FROM channels WHERE community_id=$1 AND id=$2",
+        )
+        .bind(community.as_uuid())
+        .bind(channel)
+        .fetch_one(&pool)
+        .await
+        .unwrap()
+    };
+    assert_eq!(
+        apply(&db, community, &actor, mark(channel, None, &root)).await,
+        IntentOutcome::Applied
+    );
+    let caught_up = last(pool.clone()).await;
+    assert!(caught_up.is_some(), "the fixture's message set it");
+    assert!(!sidebar(&db, community, &actor).await.unread);
+
+    // A thread reply is not on the timeline: the channel stays read.
+    reply(
+        &db,
+        community,
+        channel,
+        &root,
+        &other,
+        base + 1,
+        vec![],
+        false,
+    )
+    .await;
+    assert_eq!(last(pool.clone()).await, caught_up);
+    assert!(!sidebar(&db, community, &actor).await.unread);
+
+    // A reaction (not an eligible kind) doesn't move it; a broadcast reply,
+    // which shows on the timeline, does.
+    post_kind(&db, community, channel, &other, 7, base + 2, vec![]).await;
+    assert_eq!(last(pool.clone()).await, caught_up);
+    let broadcast = reply(
+        &db,
+        community,
+        channel,
+        &root,
+        &other,
+        base + 3,
+        vec![],
+        true,
+    )
+    .await;
+    assert!(last(pool.clone()).await > caught_up);
+    assert!(sidebar(&db, community, &actor).await.unread);
+
+    // Deleting the only unread timeline message leaves the channel read: the
+    // probe confirms what the arrival time suggests.
+    sqlx::query("UPDATE events SET deleted_at=clock_timestamp() WHERE community_id=$1 AND id=$2")
+        .bind(community.as_uuid())
+        .bind(broadcast.id.as_bytes().as_slice())
+        .execute(&pool)
+        .await
+        .unwrap();
+    assert!(!sidebar(&db, community, &actor).await.unread);
+}
+
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn top_level_message_with_depth_zero_metadata_marks_read_and_follows() {

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

@@ -84,8 +84,11 @@ impl Db {
         // 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);
+        // - unread: nothing when the channel's last timeline arrival is at
+        //   or before the position (one comparison, so a caught-up reader
+        //   never walks replies); otherwise the first timeline message past
+        //   it (idx_events_community_channel_created), which rules out a
+        //   deleted latest message;
         // - mentions: the actor's mentions in that scope that arrived past
         //   the position (idx_event_mentions_scope_received);
         // - read_through_id: the newest timeline message, by author time, that
@@ -100,6 +103,7 @@ impl Db {
             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,
+                    c.last_timeline_received_at,
                     GREATEST(cf.through_timestamp, cm.joined_at, s.started) AS position,
                     GREATEST(cf.threads_through_timestamp, cm.joined_at, s.started) AS threads_floor
                 FROM channel_members cm JOIN channels c
@@ -114,7 +118,7 @@ impl Db {
                 ORDER BY c.id LIMIT $5
              )
              SELECT r.id, r.name, r.channel_type, r.archived, r.hidden,
-                EXISTS (SELECT 1 FROM events e {EVENT_THREAD_ROW}
+                COALESCE(r.last_timeline_received_at > r.position, false) AND EXISTS (SELECT 1 FROM events e {EVENT_THREAD_ROW}
                     WHERE e.community_id=$1 AND e.channel_id=r.id
                         AND {UNREAD_EVENT} AND {ON_TIMELINE}) AS unread,
                 (SELECT count(*) FROM event_mentions m JOIN events e ON e.community_id=$1

docs/buzz-v1-read-state.md

@@ -215,8 +215,11 @@ device-local.
 
 - 20 sidebar rows, 100 intents, 20 contexts / 100 selectors per request.
 - 64 KiB write body; 16 KiB context URL; 1 MiB serialized API response.
-- `unread` is an existence probe that stops at the first message past the
-  position, so a caught-up reader examines almost nothing. Counting is forward
+- A channel whose last timeline arrival (`channels.last_timeline_received_at`)
+  is at or before the position is read with one comparison, without walking
+  past replies. Otherwise `unread` is an existence probe that stops at the
+  first message past the position, which also rules out a deleted latest
+  message. Counting is forward
   from the position: work grows with what is unread, not with history.
 - Mention counts walk only your mentions in that scope past the position
   (`idx_event_mentions_scope_received`).
@@ -228,15 +231,19 @@ device-local.
   containment, not a production capacity claim.
 
 Apply migration 0056 (or the equivalent desired schema). It creates two
-private tables, adds `thread_metadata.last_reply_received_at` and
+private tables, adds `thread_metadata.last_reply_received_at`,
+`channels.last_timeline_received_at` and
 `event_mentions.received_at`/`root_id` (NULL for rows from before it, which no
 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.
+read. Those mentions never count, an author's own message from such a
+process can show unread to them until they next read the scope, and a
+channel or thread whose only new messages came through such a process shows
+read until its next message through new code (the arrival columns are not
+updated).
 
 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,

migrations/0056_personal_read_state.sql

@@ -41,6 +41,14 @@ CREATE TABLE personal_read_frontiers (
 -- with nothing new since the reader's position without reading replies.
 ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ;
 
+-- Arrival time of a channel's latest eligible timeline message (top-level or
+-- broadcast reply): the sidebar answers "anything new?" for a caught-up
+-- reader with one comparison instead of walking past replies. Unindexed so
+-- updates stay HOT. NULL for channels with no timeline message since this
+-- migration, which no position can be behind: every position is floored at
+-- a started_at set after it.
+ALTER TABLE channels ADD COLUMN last_timeline_received_at TIMESTAMPTZ;
+
 -- Arrival time and read scope of each mention, so the sidebar counts a
 -- reader's unread mentions in one channel timeline or thread from its
 -- position. root_id is the thread a reply belongs to, NULL when the mention

schema/schema.sql

@@ -101,6 +101,8 @@ CREATE TABLE channels (
     participant_hash BYTEA,
     ttl_seconds     INT,
     ttl_deadline    TIMESTAMPTZ,
+    -- Arrival of the latest eligible timeline message; see 0056.
+    last_timeline_received_at TIMESTAMPTZ,
     PRIMARY KEY (community_id, id),
     CONSTRAINT chk_channels_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid)
 );