feat(read-state): server-owned read state for every client (buzz_v1)

crates/buzz-core/src/kind.rs

@@ -182,6 +182,7 @@ pub const RESULT_GATED_KINDS: &[u32] = &[
     KIND_AGENT_TURN_METRIC,
     KIND_MEMBER_ADDED_NOTIFICATION,
     KIND_MEMBER_REMOVED_NOTIFICATION,
+    KIND_READ_STATE_CHANGED,
 ];
 
 /// Kinds whose stored events have `#p`-bound read access — readable only by
@@ -201,6 +202,7 @@ pub const RESULT_GATED_KINDS: &[u32] = &[
 /// storage-layer search defense does not apply to them.
 pub const P_GATED_KINDS: &[u32] = &[
     KIND_AGENT_OBSERVER_FRAME,
+    KIND_READ_STATE_CHANGED,
     KIND_MEMBER_ADDED_NOTIFICATION,
     KIND_MEMBER_REMOVED_NOTIFICATION,
     KIND_GIFT_WRAP,
@@ -515,6 +517,11 @@ pub const KIND_TYPING_INDICATOR: u32 = 20002;
 pub const KIND_AGENT_PRESENCE: u32 = 20003;
 /// Ephemeral: owner-scoped encrypted agent observer telemetry and control frame.
 pub const KIND_AGENT_OBSERVER_FRAME: u32 = 24200;
+/// Ephemeral, relay-signed: the signer's `/buzz/v1` read state changed. Its
+/// `p` tag names the actor, and only that actor's connections receive it; a
+/// `nonce` tag keeps every ping's ID distinct. Content is
+/// `{"read_revision_from":n|null,"read_revision":n|null,"channels":[uuid],"threads":[{"channel_id","root_id"}]}`.
+pub const KIND_READ_STATE_CHANGED: u32 = 24300;
 /// Ephemeral: huddle emoji reaction burst. Channel-scoped to the ephemeral
 /// huddle channel with an `h` tag; never stored in the timeline.
 pub const KIND_HUDDLE_REACTION: u32 = 24810;

crates/buzz-db/src/runtime/migration.rs

@@ -776,7 +776,17 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 73);
+        assert_eq!(migrations.len(), 75);
+        assert_eq!(migrations[74].version, 75);
+        assert!(migrations[74]
+            .sql
+            .as_str()
+            .contains("ADD COLUMN revision BIGINT NOT NULL DEFAULT 0"));
+        assert_eq!(migrations[73].version, 74);
+        assert!(migrations[73]
+            .sql
+            .as_str()
+            .contains("idx_events_read_timeline"));
         assert_eq!(migrations[72].version, 73);
         assert_eq!(migrations[71].version, 72);
         assert_eq!(migrations[70].version, 71);
@@ -2425,16 +2435,41 @@ mod postgres_tests {
             .sql
             .as_str();
         assert!(through_message.contains("DROP COLUMN threads_through_timestamp"));
+        // 0074 adds the sticky thread unfollow and follow liveness.
+        let unfollowed_column = "unfollowed boolean not null default false \
+            check (not unfollowed or root_id <> ''::bytea), \
+            explicit boolean not null default false, \
+            active_at timestamptz not null default now(), ";
+        let access_paths = MIGRATOR
+            .iter()
+            .find(|m| m.version == 74)
+            .expect("read-state access path migration")
+            .sql
+            .as_str();
+        assert!(access_paths.contains(
+            "$d$BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea)$d$"
+        ));
+        assert!(access_paths
+            .contains("('personal_read_frontiers', 'explicit', 'BOOLEAN NOT NULL DEFAULT false')"));
+        assert!(access_paths.contains(
+            "('personal_read_frontiers', 'active_at', 'TIMESTAMPTZ NOT NULL DEFAULT now()')"
+        ));
         for (table, definition) in personal.tables {
             let in_schema = schema.tables.get(&table).map(|schema_definition| {
                 if table == "personal_read_accounts" {
-                    schema_definition.replacen(started_at_column, "", 1)
+                    schema_definition
+                        .replacen(started_at_column, "", 1)
+                        .replacen("revision bigint not null default 0, ", "", 1)
                 } else {
                     schema_definition.clone()
                 }
             });
             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
             };

crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs

@@ -289,20 +289,18 @@ 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. 0074
+    // 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, 73).await;
     sqlx::query(insert)
         .bind(community)
         .bind(channel)
         .bind(vec![4_u8; 32])
         .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)
@@ -316,6 +314,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!(

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

@@ -456,6 +456,10 @@ 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, 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?;
     }
     Ok(result)
@@ -493,8 +497,8 @@ async fn insert_event_on(
         .await?;
     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
         "#,
     )
@@ -510,6 +514,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?;
 
@@ -1876,14 +1881,25 @@ 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),
+    };
 
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
 
     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
         "#,
     )
@@ -1899,6 +1915,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?;
 
@@ -1913,6 +1930,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;
 
@@ -1922,8 +1945,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
                 "#,
             )
@@ -1937,6 +1960,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?;