Live thread-summary push: badge counts update on reply ingest

crates/buzz-relay/src/handlers/ingest.rs

@@ -2047,6 +2047,20 @@ async fn ingest_event_inner(
         }
     }
 
+    // A freshly inserted reply changed its thread's counters (updated in the
+    // same transaction as the insert) — push a fresh relay-signed 39005 so
+    // subscribed clients can update badge counts without refetching the head
+    // window. Page responses recompute summaries independently, so this is
+    // fan-out-only and best-effort.
+    if let Some(meta) = &thread_meta {
+        crate::handlers::side_effects::emit_live_thread_summary(
+            tenant,
+            state,
+            meta.channel_id,
+            meta.root_event_id.clone(),
+        );
+    }
+
     let pubkey_hex = auth.pubkey().to_hex();
     // Spec WriteInsert (line 514) / WriteInsertGlobal (line 559) /
     // WriteDuplicate (line 606): emit the abstract write at the trailing

crates/buzz-relay/src/handlers/side_effects.rs

@@ -11,7 +11,9 @@ use buzz_core::kind::{
     KIND_GIT_REPO_ANNOUNCEMENT, KIND_IA_ARCHIVED, KIND_IA_ARCHIVED_LIST, KIND_IA_UNARCHIVED,
     KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_NIP29_GROUP_ADMINS,
     KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, KIND_NIP43_MEMBERSHIP_LIST, KIND_REACTION,
+    KIND_THREAD_SUMMARY,
 };
+use buzz_core::StoredEvent;
 use buzz_db::channel::MemberRole;
 
 use super::event::dispatch_persistent_event;
@@ -673,6 +675,108 @@ pub async fn emit_system_message(
     Ok(())
 }
 
+/// Sign and fan out a fresh relay-signed `kind:39005` thread-summary overlay
+/// for `root_id` after a thread mutation (reply insert or threaded delete).
+///
+/// Fan-out only — never stored. Channel-window pages recompute summaries from
+/// `thread_metadata` on every fetch (`api/bridge.rs`), so a persisted copy
+/// would only add staleness; this live emit exists purely so subscribed
+/// clients can update badge counts without refetching the head window. The
+/// counts are re-read from `thread_metadata` post-commit rather than
+/// incremented, so the emitted summary is exact even under concurrent
+/// replies/deletes (newest `created_at` wins client-side).
+///
+/// Spawned: runs after the triggering write committed and must not add
+/// latency to the ingest acknowledgement, mirroring
+/// `dispatch_persistent_event`.
+pub fn emit_live_thread_summary(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
+    channel_id: Uuid,
+    root_id: Vec<u8>,
+) {
+    let tenant = tenant.clone();
+    let state = Arc::clone(state);
+    tokio::spawn(async move {
+        let summary = match state
+            .db
+            .get_thread_summary(tenant.community(), &root_id)
+            .await
+        {
+            Ok(Some(summary)) => summary,
+            // Root has no thread row — nothing to summarize (e.g., the root
+            // itself was just deleted).
+            Ok(None) => return,
+            Err(e) => {
+                warn!(
+                    root = %hex::encode(&root_id),
+                    "live thread summary lookup failed: {e}"
+                );
+                return;
+            }
+        };
+
+        let root_hex = hex::encode(&root_id);
+        // Same tags/content shape as the channel-window page overlay in
+        // `api/bridge.rs` — one contract, two delivery doors.
+        let content = serde_json::json!({
+            "reply_count": summary.reply_count,
+            "descendant_count": summary.descendant_count,
+            "last_reply_at": summary.last_reply_at.map(|t| t.timestamp()),
+            "participants": summary.participants.iter().map(hex::encode).collect::<Vec<_>>(),
+        });
+        let tags = [
+            Tag::parse(["e", &root_hex]),
+            Tag::parse(["d", &root_hex]),
+            Tag::parse(["h", &channel_id.to_string()]),
+        ];
+        let mut parsed = Vec::with_capacity(tags.len());
+        for tag in tags {
+            match tag {
+                Ok(tag) => parsed.push(tag),
+                Err(e) => {
+                    warn!(root = %root_hex, "live thread summary tag failed: {e}");
+                    return;
+                }
+            }
+        }
+        let event = match EventBuilder::new(
+            Kind::Custom(KIND_THREAD_SUMMARY as u16),
+            content.to_string(),
+        )
+        .tags(parsed)
+        .sign_with_keys(&state.relay_keypair)
+        {
+            Ok(event) => event,
+            Err(e) => {
+                warn!(root = %root_hex, "live thread summary sign failed: {e}");
+                return;
+            }
+        };
+
+        // Redis before local fan-out so subscribers on other relay pods
+        // receive it too, matching `dispatch_persistent_event`.
+        state.mark_local_event(tenant.community(), &event.id);
+        if let Err(e) = state
+            .pubsub
+            .publish_event(&tenant, EventTopic::Channel(channel_id), &event)
+            .await
+        {
+            state
+                .local_event_ids
+                .invalidate(&(tenant.community(), event.id.to_bytes()));
+            warn!(root = %root_hex, "live thread summary Redis publish failed: {e}");
+        }
+        let stored = StoredEvent::new(event, Some(channel_id));
+        crate::handlers::event::fan_out_event_to_local_subscribers(
+            &state,
+            tenant.community(),
+            &stored,
+        )
+        .await;
+    });
+}
+
 /// Emit a relay-signed membership notification event stored globally (channel_id = None).
 ///
 /// kind:44100 = member added, kind:44101 = member removed.
@@ -1485,6 +1589,12 @@ async fn handle_delete_event_side_effect(
         return Ok(()); // No-op: skip system message to avoid false audit records.
     }
 
+    // Thread counters were decremented in the same transaction — push a fresh
+    // relay-signed 39005 so live badge counts also count *down*.
+    if let Some(root_id) = root_id {
+        emit_live_thread_summary(tenant, state, channel_id, root_id);
+    }
+
     let actor_hex = hex::encode(event.pubkey.to_bytes());
     emit_system_message(
         tenant,
@@ -1968,6 +2078,12 @@ async fn handle_standard_deletion_event(
             continue;
         }
 
+        // Thread counters were decremented in the same transaction — push a
+        // fresh relay-signed 39005 so live badge counts also count *down*.
+        if let (Some(root_id), Some(channel_id)) = (root_id, target_event.channel_id) {
+            emit_live_thread_summary(tenant, state, channel_id, root_id);
+        }
+
         if u32::from(target_event.event.kind.as_u16()) == KIND_REACTION {
             // Try by reaction_event_id first; fall back to tuple-based removal
             // if the backfill was missed (set_reaction_event_id is best-effort).

crates/buzz-test-client/tests/e2e_relay.rs

@@ -2128,3 +2128,93 @@ async fn test_private_channel_member_cannot_grant_admin() {
     owner_client.disconnect().await.expect("disconnect owner");
     member_client.disconnect().await.expect("disconnect member");
 }
+
+/// Live badge counts: every thread mutation pushes a fresh relay-signed
+/// kind:39005 recount to channel subscribers — a reply counts up, deleting
+/// that reply counts back down — without any window refetch.
+#[tokio::test]
+#[ignore]
+async fn test_reply_ingest_pushes_live_thread_summary() {
+    let url = relay_url();
+    let keys = Keys::generate();
+    let channel = create_test_channel(&keys).await;
+    let mut client = BuzzTestClient::connect(&url, &keys).await.expect("connect");
+
+    // Root message for the thread — built locally so we keep its id.
+    let root = EventBuilder::new(Kind::Custom(9), "thread root")
+        .tags([Tag::parse(["h", channel.as_str()]).unwrap()])
+        .sign_with_keys(&keys)
+        .expect("sign root");
+    let root_id = root.id;
+    let ok = client.send_event(root).await.expect("send root");
+    assert!(ok.accepted, "root rejected: {}", ok.message);
+
+    // Live subscription shaped like the desktop window-store one: channel
+    // scope with 39005 in kinds.
+    let sid = sub_id("live-summary");
+    let filter = Filter::new()
+        .kind(Kind::Custom(39005))
+        .custom_tags(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
+    client
+        .subscribe(&sid, vec![filter])
+        .await
+        .expect("subscribe");
+    client
+        .collect_until_eose(&sid, Duration::from_secs(5))
+        .await
+        .expect("EOSE");
+
+    async fn recv_summary(client: &mut BuzzTestClient) -> nostr::Event {
+        loop {
+            match client
+                .recv_event(Duration::from_secs(5))
+                .await
+                .expect("recv 39005")
+            {
+                RelayMessage::Event { event, .. } if event.kind == Kind::Custom(39005) => {
+                    return *event;
+                }
+                _ => continue,
+            }
+        }
+    }
+
+    // Reply → pushed summary counts up.
+    let h_tag = Tag::parse(["h", channel.as_str()]).unwrap();
+    let root_tag = Tag::parse(["e", &root_id.to_hex(), "", "reply"]).unwrap();
+    let reply = EventBuilder::new(Kind::Custom(9), "first reply")
+        .tags([h_tag, root_tag])
+        .sign_with_keys(&keys)
+        .expect("sign reply");
+    let reply_id = reply.id;
+    let ok = client.send_event(reply).await.expect("send reply");
+    assert!(ok.accepted, "reply rejected: {}", ok.message);
+
+    let summary = recv_summary(&mut client).await;
+    let root_tag_val = summary
+        .tags
+        .iter()
+        .find(|t| t.as_slice().first().map(String::as_str) == Some("e"))
+        .and_then(|t| t.content().map(str::to_string))
+        .expect("summary carries root e-tag");
+    assert_eq!(root_tag_val, root_id.to_hex(), "summary targets the root");
+    let content: serde_json::Value = serde_json::from_str(&summary.content).expect("JSON");
+    assert_eq!(content["reply_count"], 1, "reply counted up: {content}");
+
+    // Delete the reply → pushed summary counts back down.
+    let delete = EventBuilder::new(Kind::Custom(5), "")
+        .tags([
+            Tag::parse(["e", &reply_id.to_hex()]).unwrap(),
+            Tag::parse(["h", channel.as_str()]).unwrap(),
+        ])
+        .sign_with_keys(&keys)
+        .expect("sign delete");
+    let ok = client.send_event(delete).await.expect("send delete");
+    assert!(ok.accepted, "delete rejected: {}", ok.message);
+
+    let summary = recv_summary(&mut client).await;
+    let content: serde_json::Value = serde_json::from_str(&summary.content).expect("JSON");
+    assert_eq!(content["reply_count"], 0, "reply counted down: {content}");
+
+    client.disconnect().await.expect("disconnect");
+}

desktop/src/features/channels/ui/ChannelScreen.tsx

@@ -146,10 +146,9 @@ export function ChannelScreen({
     string | null
   >(null);
   const [editTargetId, setEditTargetId] = React.useState<string | null>(null);
-  // Thread panel state is URL-backed, but router navigation is intentionally
-  // deferred out of the click handler. Keep a tiny optimistic override so the
-  // auxiliary pane can open/close in the urgent render, then let the URL-backed
-  // state hydrate the real thread contents when it catches up.
+  // Thread panel state is URL-backed, but router navigation is deferred out
+  // of the click handler; a tiny optimistic override lets the auxiliary pane
+  // open/close in the urgent render before the URL-backed state catches up.
   const [optimisticOpenThreadHeadId, setOptimisticOpenThreadHeadId] =
     React.useState<string | null | undefined>(undefined);
   const clearOptimisticThreadOverride = React.useCallback(() => {
@@ -206,16 +205,14 @@ export function ChannelScreen({
     if (!activeChannelId || activeChannel?.isMember === false) {
       return;
     }
-    // Passive channel-open: advance the marker to the newest top-level message
-    // only (NIP-RS Option 1). Opening a channel clears the main timeline while
-    // leaving thread badges and Home inbox thread activity intact until each
-    // thread itself is read.
+    // Passive channel-open (NIP-RS Option 1): advance the marker to the newest
+    // top-level message only, clearing the main timeline while thread badges
+    // and Home inbox thread activity stay intact until each thread is read.
     markChannelRead(activeChannelId, activeReadAt, { topLevelOnly: true });
   }, [activeChannel?.isMember, activeChannelId, activeReadAt, markChannelRead]);
-  // Install the NIP-RS parent resolver. Active `thread:` and `msg:` contexts
-  // belong to this channel; folding messages to the channel (never another
-  // message) preserves ancestor/descendant isolation while channel reads still
-  // cover top-level history. Clear on leave so the parent cannot go stale.
+  // Install the NIP-RS parent resolver. Active `thread:`/`msg:` contexts fold
+  // to this channel (never another message), preserving ancestor/descendant
+  // isolation while channel reads cover top-level history. Clear on leave.
   React.useEffect(() => {
     if (!activeChannelId) {
       setContextParentResolver(null);
@@ -550,9 +547,8 @@ export function ChannelScreen({
         : undefined,
     [activeChannel, handleToggleReaction],
   );
-  // The menu actions are typed (message) => void; the per-message read-state
-  // handlers key off the message id (message + subtree). Adapt at the seam so
-  // the handlers stay id-based and the menu stays message-based.
+  // The menu actions are typed (message) => void; the read-state handlers
+  // key off the message id (message + subtree). Adapt at the seam.
   const handleMessageMarkUnread = React.useCallback(
     (message: TimelineMessage) => handleMarkMessageUnread(message.id),
     [handleMarkMessageUnread],
@@ -607,10 +603,9 @@ export function ChannelScreen({
   } = useChannelAgentSessions({
     activeChannel,
     activeChannelId,
-    // The agent list comes from three queries; treat it as loaded only once
-    // none of them are in their initial fetch, so a channel with genuinely
-    // zero agents can still auto-close a stale agentSession param. A disabled
-    // query (e.g. no active channel) reports isLoading=false, which is fine.
+    // Loaded only once none of the three agent queries are in their initial
+    // fetch, so a channel with genuinely zero agents still auto-closes a stale
+    // agentSession param (a disabled query reports isLoading=false — fine).
     agentsLoaded:
       !channelMembersQuery.isLoading &&
       !managedAgentsQuery.isLoading &&
@@ -640,10 +635,10 @@ export function ChannelScreen({
       setThreadReplyTargetId,
       setThreadScrollTargetId,
     });
-  // `data !== undefined` is not "loaded": the cache is seeded early by stale
-  // placeholders and the live subscription. Wait for the history fetch to settle.
-  // Latch loaded per channel so a later background refetch can't flip back to
-  // the skeleton — that re-flip is the "skeleton bouncing up and down" on entry.
+  // `data !== undefined` is not "loaded" (the cache is seeded early by stale
+  // placeholders and the live subscription); wait for the history fetch to
+  // settle, latched per channel so a background refetch can't re-flip to the
+  // skeleton (the "skeleton bouncing up and down" on entry).
   const settledChannelIdRef = React.useRef<string | null>(null);
   const hasSettledThisChannel =
     activeChannelId !== null && settledChannelIdRef.current === activeChannelId;
@@ -667,8 +662,7 @@ export function ChannelScreen({
     );
   settledChannelIdRef.current = settledChannelId;
   // Panel identity (thread/profile/agent session) lives in the URL search
-  // params, so channel changes and back/forward traversals carry it per
-  // history entry — only the local ephemeral targets need resetting here.
+  // params and carries per history entry — only ephemeral targets reset here.
   const resetComposerTargets = React.useCallback(
     (_channelId: string | null) => {
       setExpandedThreadReplyIds(new Set());

desktop/src/features/messages/hooks.ts

@@ -39,13 +39,18 @@ import {
   emptyChannelWindowStore,
   flattenChannelWindowEvents,
   mergeLiveChannelWindowEvent,
+  mergeLiveThreadSummary,
   replaceNewestChannelWindow,
   type ChannelWindowStore,
 } from "@/features/messages/lib/channelWindowStore";
-import { parseChannelWindowResponse } from "@/features/messages/lib/channelWindowResponse";
+import {
+  parseChannelWindowResponse,
+  parseLiveThreadSummary,
+} from "@/features/messages/lib/channelWindowResponse";
 import {
   CHANNEL_AUX_EVENT_KINDS,
   CHANNEL_TIMELINE_CONTENT_KINDS,
+  KIND_CHANNEL_THREAD_SUMMARY,
   KIND_STREAM_MESSAGE,
   KIND_SYSTEM_MESSAGE,
 } from "@/shared/constants/kinds";
@@ -318,6 +323,19 @@ export function useChannelSubscription(channel: Channel | null) {
 
   const appendMessage = useEffectEvent((event: RelayEvent) => {
     if (!channelId) return;
+    if (event.kind === KIND_CHANNEL_THREAD_SUMMARY) {
+      // Relay-pushed live badge recount — window-store overlay only, never a
+      // timeline row (mirrors the page path, where 39005 is metadata).
+      const parsed = parseLiveThreadSummary(event);
+      if (!parsed) return;
+      const windowKey = channelWindowKey(channelId);
+      const current =
+        queryClient.getQueryData<ChannelWindowStore>(windowKey) ??
+        emptyChannelWindowStore();
+      const next = mergeLiveThreadSummary(current, parsed.rootId, parsed.live);
+      if (next !== current) queryClient.setQueryData(windowKey, next);
+      return;
+    }
     const isTimelineRow = CHANNEL_TIMELINE_KINDS.has(event.kind);
     const threadReference = isTimelineRow
       ? getThreadReference(event.tags)