diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 32b7244b40..d10249704e 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/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 diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index c5485721b0..bbce616d7c 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/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, + channel_id: Uuid, + root_id: Vec, +) { + 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::>(), + }); + 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). diff --git a/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index 9cd5013a68..4b5913bf47 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/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"); +} diff --git a/desktop/src/features/channels/ui/ChannelScreen.tsx b/desktop/src/features/channels/ui/ChannelScreen.tsx index 588f42a9ec..1e5cef9323 100644 --- a/desktop/src/features/channels/ui/ChannelScreen.tsx +++ b/desktop/src/features/channels/ui/ChannelScreen.tsx @@ -146,10 +146,9 @@ export function ChannelScreen({ string | null >(null); const [editTargetId, setEditTargetId] = React.useState(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(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(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()); diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index f262d9e79e..0dc1e02b78 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/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(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) diff --git a/desktop/src/features/messages/lib/channelWindowResponse.test.mjs b/desktop/src/features/messages/lib/channelWindowResponse.test.mjs index 2e001c6651..23123ebbaa 100644 --- a/desktop/src/features/messages/lib/channelWindowResponse.test.mjs +++ b/desktop/src/features/messages/lib/channelWindowResponse.test.mjs @@ -1,6 +1,9 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { parseChannelWindowResponse } from "./channelWindowResponse.ts"; +import { + parseChannelWindowResponse, + parseLiveThreadSummary, +} from "./channelWindowResponse.ts"; function event(id, kind, createdAt, content = "", tags = []) { return { @@ -125,3 +128,44 @@ test("rejects absent or contradictory signed bounds", () => { /disagree/, ); }); + +test("parses a relay-pushed live thread summary", () => { + const rootId = "a".padEnd(64, "0"); + const push = event( + "s", + 39005, + 700, + JSON.stringify({ + reply_count: 4, + descendant_count: 6, + last_reply_at: 650, + participants: ["c".repeat(64)], + }), + [ + ["e", rootId], + ["d", rootId], + ], + ); + const parsed = parseLiveThreadSummary(push); + assert.equal(parsed.rootId, rootId); + assert.equal(parsed.live.createdAt, 700); + assert.deepEqual(parsed.live.summary, { + replyCount: 4, + descendantCount: 6, + lastReplyAt: 650, + participantPubkeys: ["c".repeat(64)], + }); +}); + +test("drops malformed or mistargeted live thread summaries", () => { + const rootId = "a".padEnd(64, "0"); + // Wrong kind. + assert.equal(parseLiveThreadSummary(event("x", 9, 700, "hi")), null); + // No e-tag root. + assert.equal(parseLiveThreadSummary(event("s", 39005, 700, "{}")), null); + // Unparseable content only skips one refresh instead of throwing. + assert.equal( + parseLiveThreadSummary(event("s", 39005, 700, "not json", [["e", rootId]])), + null, + ); +}); diff --git a/desktop/src/features/messages/lib/channelWindowResponse.ts b/desktop/src/features/messages/lib/channelWindowResponse.ts index 7fc41ca995..4ff2c8f513 100644 --- a/desktop/src/features/messages/lib/channelWindowResponse.ts +++ b/desktop/src/features/messages/lib/channelWindowResponse.ts @@ -9,6 +9,7 @@ import type { ChannelWindowCursor, ChannelWindowPage, ChannelWindowThreadSummary, + LiveThreadSummary, } from "./channelWindowStore"; const CONTENT_KINDS = new Set(CHANNEL_TIMELINE_CONTENT_KINDS); @@ -38,6 +39,38 @@ function parseJson(event: RelayEvent, label: string): T { const mapCursor = (cursor: WireCursor | null): ChannelWindowCursor | null => cursor ? { createdAt: cursor.created_at, eventId: cursor.id } : null; +const mapSummary = (payload: SummaryPayload): ChannelWindowThreadSummary => ({ + replyCount: payload.reply_count, + descendantCount: payload.descendant_count, + lastReplyAt: payload.last_reply_at, + participantPubkeys: payload.participants, +}); + +/** + * Parse a relay-pushed live `39005` into its root id and summary, or null for + * anything that is not a well-formed thread summary. Live-path counterpart of + * the page parsing below — same wire contract, delivered by subscription. + */ +export function parseLiveThreadSummary( + event: RelayEvent, +): { rootId: string; live: LiveThreadSummary } | null { + if (event.kind !== KIND_CHANNEL_THREAD_SUMMARY) return null; + const rootId = targetId(event, "e"); + if (!rootId) return null; + try { + return { + rootId, + live: { + summary: mapSummary(JSON.parse(event.content) as SummaryPayload), + createdAt: event.created_at, + }, + }; + } catch { + // A malformed live overlay only skips one badge refresh — drop it. + return null; + } +} + function expectedBoundsKey( channelId: string, startCursor: ChannelWindowCursor | null, @@ -68,12 +101,7 @@ export function parseChannelWindowResponse( const row = rootId ? rowById.get(rootId) : undefined; if (!row) continue; const payload = parseJson(event, "thread summary"); - row.thread = { - replyCount: payload.reply_count, - descendantCount: payload.descendant_count, - lastReplyAt: payload.last_reply_at, - participantPubkeys: payload.participants, - }; + row.thread = mapSummary(payload); } const boundsEvents = events.filter( diff --git a/desktop/src/features/messages/lib/channelWindowStore.test.mjs b/desktop/src/features/messages/lib/channelWindowStore.test.mjs index b30804613b..91749ecf49 100644 --- a/desktop/src/features/messages/lib/channelWindowStore.test.mjs +++ b/desktop/src/features/messages/lib/channelWindowStore.test.mjs @@ -2,9 +2,11 @@ import assert from "node:assert/strict"; import test from "node:test"; import { appendOlderChannelWindow, + channelWindowThreadSummaries, emptyChannelWindowStore, flattenChannelWindowEvents, mergeLiveChannelWindowEvent, + mergeLiveThreadSummary, replaceNewestChannelWindow, } from "./channelWindowStore.ts"; @@ -269,3 +271,74 @@ test("flattening dedupes aux closure events returned on adjacent pages", () => { 1, ); }); + +const summary = (replyCount) => ({ + replyCount, + descendantCount: replyCount, + lastReplyAt: 500, + participantPubkeys: ["c".repeat(64)], +}); +const live = (replyCount, createdAt) => ({ + summary: summary(replyCount), + createdAt, +}); + +test("live thread summary overlays the page summary for its root", () => { + const root = event("a", 100); + const first = page(null, [root], { hasMore: false }); + first.rows[0].thread = summary(1); + let store = replaceNewestChannelWindow(emptyChannelWindowStore(), first); + store = mergeLiveThreadSummary(store, root.id, live(2, 200)); + assert.equal(channelWindowThreadSummaries(store).get(root.id).replyCount, 2); +}); + +test("live thread summary gives a badge to roots with no page summary", () => { + const store = mergeLiveThreadSummary( + replaceNewestChannelWindow( + emptyChannelWindowStore(), + page(null, [event("a", 100)], { hasMore: false }), + ), + event("a", 100).id, + live(1, 200), + ); + const summaries = channelWindowThreadSummaries(store); + assert.equal(summaries.get(event("a", 100).id).replyCount, 1); +}); + +test("stale live thread summaries never roll a newer one back", () => { + const rootId = event("a", 100).id; + const store = mergeLiveThreadSummary( + emptyChannelWindowStore(), + rootId, + live(3, 300), + ); + // Same-timestamp and older pushes are both ignored; identity proves no churn. + assert.equal(mergeLiveThreadSummary(store, rootId, live(2, 300)), store); + assert.equal(mergeLiveThreadSummary(store, rootId, live(2, 250)), store); + assert.equal(channelWindowThreadSummaries(store).get(rootId).replyCount, 3); +}); + +test("head refetch clears live summaries in favor of the fresh snapshot", () => { + const root = event("a", 100); + let store = mergeLiveThreadSummary( + emptyChannelWindowStore(), + root.id, + live(7, 200), + ); + const refreshed = page(null, [root], { hasMore: false }); + refreshed.rows[0].thread = summary(4); + store = replaceNewestChannelWindow(store, refreshed); + assert.equal(channelWindowThreadSummaries(store).get(root.id).replyCount, 4); +}); + +test("live summaries survive scrollback pages and still win for their root", () => { + const head = event("b", 200); + const older = event("a", 100); + const first = page(null, [head]); + let store = replaceNewestChannelWindow(emptyChannelWindowStore(), first); + store = mergeLiveThreadSummary(store, older.id, live(5, 300)); + const tail = page(first.nextCursor, [older], { hasMore: false }); + tail.rows[0].thread = summary(4); + store = appendOlderChannelWindow(store, tail); + assert.equal(channelWindowThreadSummaries(store).get(older.id).replyCount, 5); +}); diff --git a/desktop/src/features/messages/lib/channelWindowStore.ts b/desktop/src/features/messages/lib/channelWindowStore.ts index 23adc5a4eb..c41d78501a 100644 --- a/desktop/src/features/messages/lib/channelWindowStore.ts +++ b/desktop/src/features/messages/lib/channelWindowStore.ts @@ -11,6 +11,11 @@ export type ChannelWindowRow = { event: RelayEvent; thread: ChannelWindowThreadSummary | null; }; +export type LiveThreadSummary = { + summary: ChannelWindowThreadSummary; + /** `created_at` of the relay 39005 that carried it — newest wins per root. */ + createdAt: number; +}; export type ChannelWindowPage = { startCursor: ChannelWindowCursor | null; rows: ChannelWindowRow[]; @@ -24,12 +29,18 @@ export type ChannelWindowStore = { liveOverlay: RelayEvent[]; /** Live structural events retained independently from frozen page closure. */ liveAux: RelayEvent[]; + /** + * Relay-pushed 39005 summaries (keyed by thread root id) not yet superseded + * by a page. A page response supersedes them for the roots it re-delivers. + */ + liveSummaries: Record; }; export const emptyChannelWindowStore = (): ChannelWindowStore => ({ pages: [], liveOverlay: [], liveAux: [], + liveSummaries: {}, }); function cursorsEqual( @@ -103,6 +114,10 @@ export function replaceNewestChannelWindow( pages: [page], liveOverlay: current.liveOverlay.filter((event) => !ids.has(event.id)), liveAux: current.liveAux.filter((event) => !auxIds.has(event.id)), + // A head refetch is the authoritative resync moment (subscribe/reconnect + // both funnel here): every retained page is replaced, so live summaries + // pinned to the old snapshot are cleared rather than diffed. + liveSummaries: {}, }; } @@ -140,6 +155,27 @@ export function appendOlderChannelWindow( }; } +/** + * Record a relay-pushed live `39005` summary. Newest `created_at` wins per + * root: the relay pushes a full recount on every thread mutation, so the + * latest push is authoritative for that root — including counting *down* + * after a delete. Retained across scrollback pages (a racing push can be + * fresher than a just-fetched page summary) and cleared only by the head + * refetch in `replaceNewestChannelWindow`. + */ +export function mergeLiveThreadSummary( + current: ChannelWindowStore, + rootId: string, + incoming: LiveThreadSummary, +): ChannelWindowStore { + const existing = current.liveSummaries[rootId]; + if (existing && existing.createdAt >= incoming.createdAt) return current; + return { + ...current, + liveSummaries: { ...current.liveSummaries, [rootId]: incoming }, + }; +} + /** * Merge a live top-level event without mutating authoritative page boundaries. * Events below the oldest loaded boundary wait for ordinary relay pagination. @@ -205,12 +241,22 @@ export function channelWindowHasMore(store: ChannelWindowStore) { return tail?.hasMore ?? false; } +/** + * Per-root thread summaries for badge rendering: authoritative page summaries + * overlaid with any fresher relay-pushed live summaries. The live overlay also + * covers roots that reached the screen outside a page (liveOverlay rows, + * refetch-reconciled rows), which have no page summary at all. + */ export function channelWindowThreadSummaries(store: ChannelWindowStore) { - return new Map( + const summaries = new Map( store.pages.flatMap((page) => page.rows.flatMap((row) => row.thread ? ([[row.event.id, row.thread]] as const) : [], ), ), ); + for (const [rootId, live] of Object.entries(store.liveSummaries)) { + summaries.set(rootId, live.summary); + } + return summaries; } diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 1298f21016..c2462d08ca 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -11,6 +11,7 @@ import { KIND_TYPING_INDICATOR, KIND_USER_STATUS, CHANNEL_EVENT_KINDS, + KIND_CHANNEL_THREAD_SUMMARY, } from "@/shared/constants/kinds"; import { getTextPayload, @@ -340,7 +341,10 @@ export class RelayClient { ) { return this.subscribe( { - kinds: [...CHANNEL_EVENT_KINDS], + // 39005 rides only this window-store subscription — not + // CHANNEL_EVENT_KINDS, whose other consumers (unread tracking, + // timeline-cache merges) must never see summary overlays. + kinds: [...CHANNEL_EVENT_KINDS, KIND_CHANNEL_THREAD_SUMMARY], "#h": [channelId], limit: 1000, since: Math.floor(Date.now() / 1_000),