diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index bb0512c4f..e578e8277 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -45,6 +45,7 @@ mod terminal_runtime; mod terminal_transport; #[cfg(target_os = "macos")] mod tray_menu; +mod unread_catch_up; mod util; #[cfg(target_os = "linux")] pub mod webkit_rendering; @@ -716,6 +717,7 @@ pub fn run() { discover_backend_providers, probe_backend_provider, persona_catalog::fetch_persona_catalog, + unread_catch_up::unread_catch_up, list_personas, create_persona, update_persona, diff --git a/desktop/src-tauri/src/persona_catalog_tests.rs b/desktop/src-tauri/src/persona_catalog_tests.rs index 0532660b4..d3175ef98 100644 --- a/desktop/src-tauri/src/persona_catalog_tests.rs +++ b/desktop/src-tauri/src/persona_catalog_tests.rs @@ -190,3 +190,46 @@ fn exact_tags_reject_duplicates_and_extra_fields() { assert_eq!(coordinate_tag(&extended, "d").as_deref(), Some("reviewer")); assert_eq!(exact_tag(&extended, "shared"), None); } + +/// Pins the serialized DTO output against the renderer's catalog contract. +/// The Tauri generic is only a TypeScript assertion; serde's bytes are the +/// actual boundary, so populate every optional field and compare the value. +#[test] +fn serialized_catalog_matches_the_typescript_contract() { + let publication = PersonaCatalogPublication { + event_id: "ev1".into(), + owner_pubkey: "owner".into(), + source_persona_id: "persona-1".into(), + created_at: 42, + agent: CatalogAgentProjection { + display_name: "Ada".into(), + avatar_url: Some("https://example.com/a.png".into()), + system_prompt: "be kind".into(), + runtime: Some("acp".into()), + model: Some("m1".into()), + provider: Some("p1".into()), + name_pool: vec!["Ada".into(), "Lin".into()], + respond_to: Some("mentions".into()), + parallelism: Some(2), + }, + }; + let actual = serde_json::to_value(vec![publication]).unwrap(); + let expected = serde_json::json!([{ + "eventId": "ev1", + "ownerPubkey": "owner", + "sourcePersonaId": "persona-1", + "createdAt": 42, + "agent": { + "displayName": "Ada", + "avatarUrl": "https://example.com/a.png", + "systemPrompt": "be kind", + "runtime": "acp", + "model": "m1", + "provider": "p1", + "namePool": ["Ada", "Lin"], + "respondTo": "mentions", + "parallelism": 2, + }, + }]); + assert_eq!(actual, expected); +} diff --git a/desktop/src-tauri/src/unread_catch_up.rs b/desktop/src-tauri/src/unread_catch_up.rs new file mode 100644 index 000000000..c7de5dae8 --- /dev/null +++ b/desktop/src-tauri/src/unread_catch_up.rs @@ -0,0 +1,647 @@ +//! Batched native unread catch-up. +//! +//! The renderer supplies its history-derived notification membership because +//! those sets are still renderer-owned until the native observed-unread store +//! lands. Rust performs every channel REQ over the shared authenticated session, +//! then classifies the complete successful batch in two passes so a root learned +//! anywhere in pass one is visible everywhere in pass two. + +use std::{collections::HashSet, time::Duration}; + +use buzz_core_pkg::kind::{ + KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_HUDDLE_STARTED, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, +}; +use nostr::Event; +use serde::{Deserialize, Serialize}; +use tauri::State; +use tokio::{sync::Semaphore, task::JoinSet}; + +use crate::{app_state::AppState, native_relay_client::NativeRelayClient}; + +const CATCH_UP_LIMIT: usize = 1_000; +const ACTIVITY_LIMIT: usize = 100; +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); + +#[derive(Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct UnreadCatchUpRequest { + channels: Vec, + self_pubkey: String, + participated_root_ids: HashSet, + authored_root_ids: HashSet, + mentioned_root_ids: HashSet, + followed_root_ids: HashSet, + muted_root_ids: HashSet, + muted_channel_ids: HashSet, +} + +#[derive(Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +struct CatchUpChannel { + id: String, + #[serde(rename = "type")] + channel_type: String, + name: String, + read_at: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct UnreadCatchUpResponse { + channels: Vec, +} + +#[derive(Serialize)] +#[serde( + tag = "status", + rename_all = "camelCase", + rename_all_fields = "camelCase" +)] +enum ChannelResult { + Success { + channel_id: String, + observed_events: Vec, + max_trigger: u64, + activity_rows: Vec, + discovered: DiscoveredRoots, + }, + Error { + channel_id: String, + error: String, + }, +} + +#[derive(Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +struct ObservedUnreadEvent { + id: String, + created_at: u64, + root_id: Option, + high_priority: bool, + counts_toward_badge: bool, + counts_toward_app_badge: bool, +} + +#[derive(Clone, Serialize)] +#[serde(rename_all = "camelCase")] +struct ActivityRow { + id: String, + kind: u16, + pubkey: String, + content: String, + created_at: u64, + channel_id: String, + channel_name: String, + tags: Vec>, +} + +#[derive(Default, Serialize)] +#[serde(rename_all = "camelCase")] +struct DiscoveredRoots { + participated: Vec, + authored: Vec, + mentioned: Vec, +} + +struct FetchedChannel { + order: usize, + channel: CatchUpChannel, + events: Vec, +} + +#[derive(Clone)] +struct EventView { + id: String, + kind: u16, + pubkey: String, + content: String, + created_at: u64, + tags: Vec>, +} + +impl From for EventView { + fn from(event: Event) -> Self { + Self { + id: event.id.to_hex(), + kind: event.kind.as_u16(), + pubkey: event.pubkey.to_hex(), + content: event.content, + created_at: event.created_at.as_secs(), + tags: event + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(), + } + } +} + +#[tauri::command] +pub(crate) async fn unread_catch_up( + request: UnreadCatchUpRequest, + state: State<'_, AppState>, + relay_client: State<'_, NativeRelayClient>, +) -> Result { + let keys = state.signing_keys()?; + let owner = keys.public_key().to_hex(); + if !owner.eq_ignore_ascii_case(&request.self_pubkey) { + return Err("unread catch-up identity does not match active scope".to_string()); + } + let relay_url = crate::relay::relay_ws_url_with_override(&state); + let session = relay_client.session(relay_url.clone(), keys).await; + + let concurrency = std::sync::Arc::new(Semaphore::new(8)); + let mut pending = JoinSet::new(); + // One command replaces N renderer invokes while the shared session still + // multiplexes bounded finite REQs on one authenticated socket. + for (order, channel) in request.channels.iter().cloned().enumerate() { + let permit = concurrency + .clone() + .acquire_owned() + .await + .map_err(|error| error.to_string())?; + let session = session.clone(); + pending.spawn(async move { + let _permit = permit; + let kinds: &[u32] = if channel.channel_type == "dm" { + &[ + KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, + KIND_FORUM_POST, + KIND_FORUM_COMMENT, + KIND_HUDDLE_STARTED, + ] + } else { + &[ + KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, + KIND_FORUM_POST, + KIND_FORUM_COMMENT, + ] + }; + let filter = serde_json::json!({ + "kinds": kinds, + "#h": [channel.id], + "since": channel.read_at.map_or(0, |value| value.saturating_add(1)), + "limit": CATCH_UP_LIMIT, + }); + let result = session.fetch_events(filter, REQUEST_TIMEOUT).await; + (order, channel, result) + }); + } + + let mut fetched = Vec::new(); + let mut failures = Vec::new(); + while let Some(joined) = pending.join_next().await { + let (order, channel, result) = + joined.map_err(|error| format!("unread catch-up task failed: {error}"))?; + match result { + Ok(events) => fetched.push(FetchedChannel { + order, + channel, + events: events + .into_iter() + .take(CATCH_UP_LIMIT) + .map(EventView::from) + .collect(), + }), + Err(error) => failures.push(ChannelResult::Error { + channel_id: channel.id, + error, + }), + } + } + + fetched.sort_by_key(|item| item.order); + + let current_keys = state.signing_keys()?; + if current_keys.public_key().to_hex() != owner + || crate::relay::relay_ws_url_with_override(&state) != relay_url + { + return Err("unread catch-up scope changed while fetching".to_string()); + } + + let mut channels = classify_batch(&request, fetched); + channels.extend(failures); + Ok(UnreadCatchUpResponse { channels }) +} + +fn classify_batch( + request: &UnreadCatchUpRequest, + fetched: Vec, +) -> Vec { + let self_pubkey = request.self_pubkey.to_lowercase(); + let mut participated = request.participated_root_ids.clone(); + let mut authored = request.authored_root_ids.clone(); + let mut mentioned = request.mentioned_root_ids.clone(); + + // Pass one is deliberately global, not per-channel: notification validity + // depends on roots learned from history, while the command observes a batch. + // Deltas remain attributed to the channel that first discovered each root. + let mut discoveries = Vec::with_capacity(fetched.len()); + for item in &fetched { + let mut discovered = DiscoveredRoots::default(); + for event in &item.events { + if event.pubkey.eq_ignore_ascii_case(&self_pubkey) { + let reference = thread_reference(&event.tags); + if let Some(root_id) = reference.root_id { + if participated.insert(root_id.clone()) { + discovered.participated.push(root_id); + } + } else if authored.insert(event.id.clone()) { + discovered.authored.push(event.id.clone()); + } + } else if has_tag_value(&event.tags, "p", &self_pubkey) { + if let Some(root_id) = thread_reference(&event.tags).root_id { + if mentioned.insert(root_id.clone()) { + discovered.mentioned.push(root_id); + } + } + } + } + discoveries.push(discovered); + } + + let mut outputs = Vec::new(); + let mut all_activity = Vec::new(); + for (item, discovered) in fetched.into_iter().zip(discoveries) { + let mut observed_events = Vec::new(); + let mut activity_rows = Vec::new(); + let mut max_trigger = 0; + for event in item.events { + if event.pubkey.eq_ignore_ascii_case(&self_pubkey) + || item + .channel + .read_at + .is_some_and(|read_at| event.created_at <= read_at) + || !should_notify(&event, &self_pubkey, request, &participated, &authored) + { + continue; + } + let reference = thread_reference(&event.tags); + let broadcast = has_exact_tag(&event.tags, "broadcast", "1"); + let threaded = reference.parent_id.is_some() && !broadcast; + let high_priority = item.channel.channel_type == "dm" + || broadcast + || has_tag_value(&event.tags, "p", &self_pubkey); + max_trigger = max_trigger.max(event.created_at); + observed_events.push(ObservedUnreadEvent { + id: event.id.clone(), + created_at: event.created_at, + root_id: if broadcast { + None + } else { + reference.root_id.clone() + }, + high_priority, + counts_toward_badge: item.channel.channel_type == "dm" || threaded || high_priority, + counts_toward_app_badge: item.channel.channel_type == "dm" + || (!threaded && high_priority), + }); + if threaded { + activity_rows.push(ActivityRow { + id: event.id, + kind: event.kind, + pubkey: event.pubkey, + content: event.content, + created_at: event.created_at, + channel_id: item.channel.id.clone(), + channel_name: item.channel.name.clone(), + tags: event.tags, + }); + } + } + all_activity.extend(activity_rows.iter().cloned()); + outputs.push(( + item.channel.id, + observed_events, + max_trigger, + activity_rows, + discovered, + )); + } + + all_activity.sort_by_key(|row| row.created_at); + let mut seen = HashSet::new(); + all_activity.retain(|row| seen.insert(row.id.clone())); + if all_activity.len() > ACTIVITY_LIMIT { + all_activity.drain(..all_activity.len() - ACTIVITY_LIMIT); + } + let allowed: HashSet<_> = all_activity.into_iter().map(|row| row.id).collect(); + + outputs + .into_iter() + .map( + |(channel_id, observed_events, max_trigger, mut activity_rows, discovered)| { + activity_rows.retain(|row| allowed.contains(&row.id)); + ChannelResult::Success { + channel_id, + observed_events, + max_trigger, + activity_rows, + discovered, + } + }, + ) + .collect() +} + +struct ThreadReference { + parent_id: Option, + root_id: Option, +} + +fn thread_reference(tags: &[Vec]) -> ThreadReference { + let event_tags: Vec<_> = tags + .iter() + .filter(|tag| tag.first().is_some_and(|v| v == "e") && tag.get(1).is_some()) + .collect(); + let root = event_tags + .iter() + .find(|tag| tag.get(3).is_some_and(|v| v == "root")); + let reply = event_tags + .iter() + .rev() + .find(|tag| tag.get(3).is_some_and(|v| v == "reply")); + let Some(reply) = reply else { + return ThreadReference { + parent_id: None, + root_id: None, + }; + }; + let parent_id = reply.get(1).cloned(); + ThreadReference { + root_id: root + .and_then(|tag| tag.get(1).cloned()) + .or_else(|| parent_id.clone()), + parent_id, + } +} + +fn should_notify( + event: &EventView, + self_pubkey: &str, + request: &UnreadCatchUpRequest, + participated: &HashSet, + authored: &HashSet, +) -> bool { + if has_exact_tag(&event.tags, "broadcast", "1") || has_tag_value(&event.tags, "p", self_pubkey) + { + return true; + } + let event_channel_id = event + .tags + .iter() + .find(|tag| tag.first().is_some_and(|part| part == "h")) + .and_then(|tag| tag.get(1)); + if event_channel_id.is_some_and(|id| request.muted_channel_ids.contains(id)) { + return false; + } + let reference = thread_reference(&event.tags); + if reference.parent_id.is_none() { + return true; + } + let Some(root_id) = reference.root_id else { + return false; + }; + if request.muted_root_ids.contains(&root_id) { + return false; + } + participated.contains(&root_id) + || request.followed_root_ids.contains(&root_id) + || authored.contains(&root_id) +} + +fn has_exact_tag(tags: &[Vec], name: &str, value: &str) -> bool { + tags.iter().any(|tag| { + tag.first().is_some_and(|part| part == name) && tag.get(1).is_some_and(|part| part == value) + }) +} + +fn has_tag_value(tags: &[Vec], name: &str, value: &str) -> bool { + tags.iter().any(|tag| { + tag.first().is_some_and(|part| part == name) + && tag + .get(1) + .is_some_and(|part| part.eq_ignore_ascii_case(value)) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn event(id: &str, pubkey: &str, created_at: u64, tags: &[&[&str]]) -> EventView { + EventView { + id: id.into(), + kind: 9, + pubkey: pubkey.into(), + content: id.into(), + created_at, + tags: tags + .iter() + .map(|tag| tag.iter().map(|part| (*part).to_string()).collect()) + .collect(), + } + } + + fn request() -> UnreadCatchUpRequest { + UnreadCatchUpRequest { + channels: vec![], + self_pubkey: "self".into(), + participated_root_ids: HashSet::new(), + authored_root_ids: HashSet::new(), + mentioned_root_ids: HashSet::new(), + followed_root_ids: HashSet::new(), + muted_root_ids: HashSet::new(), + muted_channel_ids: HashSet::new(), + } + } + + #[test] + fn pass_one_history_changes_later_classification() { + let req = request(); + let channel = CatchUpChannel { + id: "ch".into(), + channel_type: "stream".into(), + name: "Ch".into(), + read_at: Some(9), + }; + let fetched = vec![FetchedChannel { + order: 0, + channel, + events: vec![ + event( + "self-reply", + "self", + 10, + &[&["e", "root", "", "reply"], &["h", "ch"]], + ), + event( + "external-reply", + "other", + 11, + &[&["e", "root", "", "reply"], &["h", "ch"]], + ), + ], + }]; + let result = classify_batch(&req, fetched); + let ChannelResult::Success { + observed_events, + discovered, + .. + } = &result[0] + else { + panic!("expected success") + }; + assert_eq!( + observed_events + .iter() + .map(|event| event.id.as_str()) + .collect::>(), + ["external-reply"] + ); + assert_eq!(discovered.participated, ["root"]); + } + + #[test] + fn same_second_marker_and_mutes_match_renderer_rules() { + let mut req = request(); + req.muted_root_ids.insert("muted".into()); + let channel = CatchUpChannel { + id: "ch".into(), + channel_type: "stream".into(), + name: "Ch".into(), + read_at: Some(10), + }; + let fetched = vec![FetchedChannel { + order: 0, + channel, + events: vec![ + event("boundary", "other", 10, &[&["h", "ch"]]), + event( + "muted", + "other", + 11, + &[&["e", "muted", "", "reply"], &["h", "ch"]], + ), + event( + "broadcast", + "other", + 12, + &[&["broadcast", "1"], &["h", "ch"]], + ), + ], + }]; + let result = classify_batch(&req, fetched); + let ChannelResult::Success { + observed_events, + max_trigger, + .. + } = &result[0] + else { + panic!("expected success") + }; + assert_eq!( + observed_events + .iter() + .map(|event| event.id.as_str()) + .collect::>(), + ["broadcast"] + ); + assert_eq!(*max_trigger, 12); + } + + /// Pins the SERIALIZED wire contract against `tauriUnreadCatchUp.ts`. + /// + /// Asserts on serde's OUTPUT, not on `ChannelResult`: the renderer never + /// sees the Rust type, it sees bytes, through an `invokeTauri` cast + /// that validates nothing. Every other test here inspects the enum before + /// serialization and the e2e bridge hand-writes the intended shape, so + /// without this nothing compares what Rust emits to what TypeScript + /// declares. + /// + /// Whole-value rather than a key list, deliberately: a key-set assertion + /// passes a mutant that drops the variant rename and emits `"Success"`, + /// which the renderer's `status === "error"` branch silently misreads. + /// Failure here means the merge loop throws on the first success row and + /// catch-up yields nothing, silently. + #[test] + fn serialized_response_matches_the_typescript_contract() { + let channels = vec![ + ChannelResult::Success { + channel_id: "ch".into(), + observed_events: vec![ObservedUnreadEvent { + id: "evt".into(), + created_at: 11, + root_id: Some("root".into()), + high_priority: true, + counts_toward_badge: true, + counts_toward_app_badge: false, + }], + max_trigger: 11, + activity_rows: vec![ActivityRow { + id: "evt".into(), + kind: 9, + pubkey: "other".into(), + content: "hi".into(), + created_at: 11, + channel_id: "ch".into(), + channel_name: "Ch".into(), + tags: vec![vec!["h".into(), "ch".into()]], + }], + discovered: DiscoveredRoots { + participated: vec!["root".into()], + authored: Vec::new(), + mentioned: Vec::new(), + }, + }, + ChannelResult::Error { + channel_id: "ch-2".into(), + error: "relay request timed out".into(), + }, + ]; + + let actual = serde_json::to_value(UnreadCatchUpResponse { channels }).unwrap(); + let expected = serde_json::json!({ + "channels": [ + { + "status": "success", + "channelId": "ch", + "observedEvents": [{ + "id": "evt", + "createdAt": 11, + "rootId": "root", + "highPriority": true, + "countsTowardBadge": true, + "countsTowardAppBadge": false, + }], + "maxTrigger": 11, + "activityRows": [{ + "id": "evt", + "kind": 9, + "pubkey": "other", + "content": "hi", + "createdAt": 11, + "channelId": "ch", + "channelName": "Ch", + "tags": [["h", "ch"]], + }], + "discovered": { + "participated": ["root"], + "authored": [], + "mentioned": [], + }, + }, + { + "status": "error", + "channelId": "ch-2", + "error": "relay request timed out", + }, + ] + }); + + assert_eq!(actual, expected); + } +} diff --git a/desktop/src/features/channels/useUnreadChannels.ts b/desktop/src/features/channels/useUnreadChannels.ts index 0464a00fe..34b2d9450 100644 --- a/desktop/src/features/channels/useUnreadChannels.ts +++ b/desktop/src/features/channels/useUnreadChannels.ts @@ -29,7 +29,6 @@ import { import { hasMentionForEvent, isHighPriorityEventForUser, - shouldNotifyForEvent, } from "@/features/notifications/lib/shouldNotify"; import type { RelayClient } from "@/shared/api/relayClientSession"; import type { Channel, RelayEvent } from "@/shared/api/types"; @@ -53,6 +52,7 @@ export { } from "@/features/channels/threadActivityStorage"; import { useObservedUnreadPersistence } from "@/features/channels/useObservedUnreadPersistence"; import { useThreadActivityPersistence } from "@/features/channels/useThreadActivityPersistence"; +import { unreadCatchUp } from "@/shared/api/tauriUnreadCatchUp"; type UseUnreadChannelsOptions = UseLiveChannelUpdatesOptions & { pubkey?: string; @@ -590,189 +590,114 @@ export function useUnreadChannels( const authoredSizeBefore = authoredRootIdsRef.current.size; const mentionedSizeBefore = mentionedRootIdsRef.current.size; - type CatchUpResult = - | { - channelId: string; - ok: true; - maxExternal: number; - unreadEvents: ObservedUnreadEvent[]; - threadReplies: ThreadActivityItem[]; - } - | { channelId: string; ok: false }; - - void Promise.all( - toFetch.map(async (channelId): Promise => { - try { - const readAt = getEffectiveTimestamp(channelId); - const channel = channels.find((c) => c.id === channelId); - // NIP-01 `since` is inclusive of `created_at >= since`. The +1 - // makes the relay-side filter strict-newer; the client-side - // `> readAt` check below is the belt to the suspenders. - const sinceParam = readAt === null ? 0 : readAt + 1; - - const events = await relayClient.fetchEvents({ - kinds: [...channelCatchUpEventKinds(channel?.channelType)], - "#h": [channelId], - since: sinceParam, - limit: CATCH_UP_LIMIT, - }); - - // Pass 1: build participation from self-authored thread replies, - // track self-authored top-level messages for author notifications, - // and capture external mentions so their threads gate a badge. - for (const event of events) { - const isSelf = - normalizedPubkey !== null && - event.pubkey.toLowerCase() === normalizedPubkey; - if (isSelf) { - const ref = getThreadReference(event.tags); - if (ref.rootId !== null) { - participatedRootIdsRef.current.add(ref.rootId); - } else { - authoredRootIdsRef.current.add(event.id); - } - } else { - recordMentionedRoot(event); - } - } - - if (normalizedPubkey !== null) { - participationStore.write( - normalizedPubkey, - participatedRootIdsRef.current, - ); - authoredStore.write(normalizedPubkey, authoredRootIdsRef.current); - } - - // Pass 2: compute maxExternal and collect thread reply activity, - // applying the notification filter to both. - let maxExternal = 0; - const unreadEvents: ObservedUnreadEvent[] = []; - const threadReplies: ThreadActivityItem[] = []; - const chType = channel?.channelType; - const chName = channel?.name ?? ""; - for (const event of events) { - if ( - normalizedPubkey !== null && - event.pubkey.toLowerCase() === normalizedPubkey - ) { - continue; - } - if (readAt !== null && event.created_at <= readAt) continue; - const eventChannelId = - event.tags.find((t) => t[0] === "h")?.[1] ?? null; - if ( - !shouldNotifyForEvent(event, normalizedPubkey ?? "", { - participatedRootIds: participatedRootIdsRef.current, - followedRootIds: options.followedRootIds ?? EMPTY_SET, - authoredRootIds: authoredRootIdsRef.current, - mutedRootIds: mutedRootIdsRef.current, - mutedChannelIds: mutedChannelIdsRef.current, - channelId: eventChannelId, - }) - ) { - continue; - } - const evtRef = getThreadReference(event.tags); - const isThreadedReply = - evtRef.parentId !== null && !isBroadcastReply(event.tags); - if (event.created_at > maxExternal) { - maxExternal = event.created_at; - } - const isHighPriority = - chType === "dm" || - (normalizedPubkey !== null && - isHighPriorityEventForUser(event, normalizedPubkey)); - unreadEvents.push( - makeObservedUnreadEvent({ - id: event.id, - createdAt: event.created_at, - rootId: resolveObservedUnreadRootId(event.tags), - highPriority: isHighPriority, - channelType: chType, - isThreadedReply, - }), - ); - if (isThreadedReply) { - threadReplies.push({ - id: event.id, - kind: event.kind, - pubkey: event.pubkey, - content: event.content, - createdAt: event.created_at, - channelId, - channelName: chName, - tags: [...event.tags], - }); - } - } - - return { - channelId, - ok: true, - maxExternal, - unreadEvents, - threadReplies, - }; - } catch { - // Transient relay failure for this channel — release the claim - // so we retry on the next effect run instead of staying stuck - // until identity reset. - return { channelId, ok: false }; - } + // Membership remains renderer-owned until E's native observed-unread store, + // so unchanged sets cross IPC on every catch-up. The five 1,000-entry + // stores bound that interim cost at roughly 332 KiB per request. Command + // arguments use a fetch body, so this cost is linear with no size cliff. + void unreadCatchUp({ + channels: toFetch.map((channelId) => { + const channel = channels.find( + (candidate) => candidate.id === channelId, + ); + return { + id: channelId, + type: channel?.channelType ?? "stream", + name: channel?.name ?? "", + readAt: getEffectiveTimestamp(channelId), + }; }), - ).then((results) => { - if (isCancelled) return; - // Guard: don't merge catch-up results into a ref whose scope has drifted - // (relay/pubkey changed while this async fetch was in flight). Use the - // observed owner's loaded-scope predicate — one scope authority, not two. - if (!observedPersistence.isScopeLoaded()) return; - let didAdvance = false; - const allThreadReplies: ThreadActivityItem[] = []; - for (const result of results) { - if (!result.ok) { - caughtUpChannelsRef.current.delete(result.channelId); - continue; - } - const { channelId, maxExternal, unreadEvents, threadReplies } = result; - allThreadReplies.push(...threadReplies); - if (unreadEvents.length > 0) { - for (const event of unreadEvents) { - recordUnreadEvent(channelId, event); + selfPubkey: normalizedPubkey ?? "", + participatedRootIds: [...participatedRootIdsRef.current], + authoredRootIds: [...authoredRootIdsRef.current], + mentionedRootIds: [...mentionedRootIdsRef.current], + followedRootIds: [...(options.followedRootIds ?? EMPTY_SET)], + mutedRootIds: [...mutedRootIdsRef.current], + mutedChannelIds: [...mutedChannelIdsRef.current], + }) + .then(({ channels: results }) => { + if (isCancelled) return; + // The command rejects if relay/pubkey scope changes in flight. Keep the + // renderer fence too: it also covers effect cleanup before merge. + if (!observedPersistence.isScopeLoaded()) return; + + let didAdvance = false; + let didDiscover = false; + const allThreadReplies: ThreadActivityItem[] = []; + for (const result of results) { + if (result.status === "error") { + // The error arm carries only this identity; releasing its claim is + // what lets a failed channel retry on the next effect run. + caughtUpChannelsRef.current.delete(result.channelId); + continue; } - didAdvance = true; - } - if (maxExternal > 0) { - const readAtNow = getEffectiveTimestamp(channelId) ?? 0; - if (maxExternal > readAtNow) { - const current = latestByChannelRef.current.get(channelId) ?? 0; - if (maxExternal > current) { - latestByChannelRef.current.set(channelId, maxExternal); + for (const rootId of result.discovered.participated) { + const before = participatedRootIdsRef.current.size; + participatedRootIdsRef.current.add(rootId); + didDiscover ||= participatedRootIdsRef.current.size !== before; + } + for (const rootId of result.discovered.authored) { + const before = authoredRootIdsRef.current.size; + authoredRootIdsRef.current.add(rootId); + didDiscover ||= authoredRootIdsRef.current.size !== before; + } + for (const rootId of result.discovered.mentioned) { + const before = mentionedRootIdsRef.current.size; + mentionedRootIdsRef.current.add(rootId); + didDiscover ||= mentionedRootIdsRef.current.size !== before; + } + allThreadReplies.push(...result.activityRows); + for (const event of result.observedEvents) { + recordUnreadEvent(result.channelId, event); + didAdvance = true; + } + if ( + result.maxTrigger > (getEffectiveTimestamp(result.channelId) ?? 0) + ) { + const current = + latestByChannelRef.current.get(result.channelId) ?? 0; + if (result.maxTrigger > current) { + latestByChannelRef.current.set( + result.channelId, + result.maxTrigger, + ); didAdvance = true; } } } - } - if (allThreadReplies.length > 0) { - const added = addThreadActivityItems( - threadActivityRef.current, - allThreadReplies, - ); - if (added.didAdd) { - threadActivityRef.current = added.items; - activityPersistence.schedule(currentActivityScope); - didAdvance = true; + + if (normalizedPubkey !== null && didDiscover) { + participationStore.write( + normalizedPubkey, + participatedRootIdsRef.current, + ); + authoredStore.write(normalizedPubkey, authoredRootIdsRef.current); + mentionedStore.write(normalizedPubkey, mentionedRootIdsRef.current); } - } - if (didAdvance) bumpLatestVersion(); - if ( - participatedRootIdsRef.current.size !== participatedSizeBefore || - authoredRootIdsRef.current.size !== authoredSizeBefore || - mentionedRootIdsRef.current.size !== mentionedSizeBefore - ) { - bumpMembershipVersion(); - } - }); + if (allThreadReplies.length > 0) { + const added = addThreadActivityItems( + threadActivityRef.current, + allThreadReplies, + ); + if (added.didAdd) { + threadActivityRef.current = added.items; + activityPersistence.schedule(currentActivityScope); + didAdvance = true; + } + } + if (didAdvance) bumpLatestVersion(); + if ( + didDiscover || + participatedRootIdsRef.current.size !== participatedSizeBefore || + authoredRootIdsRef.current.size !== authoredSizeBefore || + mentionedRootIdsRef.current.size !== mentionedSizeBefore + ) { + bumpMembershipVersion(); + } + }) + .catch(() => { + if (isCancelled) return; + for (const id of toFetch) caughtUpChannelsRef.current.delete(id); + }); return () => { isCancelled = true; diff --git a/desktop/src/shared/api/tauriUnreadCatchUp.ts b/desktop/src/shared/api/tauriUnreadCatchUp.ts new file mode 100644 index 000000000..0f0c125b0 --- /dev/null +++ b/desktop/src/shared/api/tauriUnreadCatchUp.ts @@ -0,0 +1,42 @@ +import { invokeTauri } from "@/shared/api/tauri"; +import type { ObservedUnreadEvent } from "@/features/channels/unreadChannelCounts"; +import type { ThreadActivityItem } from "@/features/channels/threadActivityStorage"; + +export type UnreadCatchUpChannel = { + id: string; + type: string; + name: string; + readAt: number | null; +}; + +export type UnreadCatchUpRequest = { + channels: UnreadCatchUpChannel[]; + selfPubkey: string; + participatedRootIds: string[]; + authoredRootIds: string[]; + mentionedRootIds: string[]; + followedRootIds: string[]; + mutedRootIds: string[]; + mutedChannelIds: string[]; +}; + +export type UnreadCatchUpChannelResult = + | { + status: "success"; + channelId: string; + observedEvents: ObservedUnreadEvent[]; + maxTrigger: number; + activityRows: ThreadActivityItem[]; + discovered: { + participated: string[]; + authored: string[]; + mentioned: string[]; + }; + } + | { status: "error"; channelId: string; error: string }; + +export function unreadCatchUp( + request: UnreadCatchUpRequest, +): Promise<{ channels: UnreadCatchUpChannelResult[] }> { + return invokeTauri("unread_catch_up", { request }); +} diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 40c129101..8a31d2ad2 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -10,6 +10,7 @@ import { handleDeleteCustomHarness, } from "./e2eBridgeCustomHarnesses.ts"; +import type { UnreadCatchUpChannelResult } from "@/shared/api/tauriUnreadCatchUp"; import { relayClient } from "@/shared/api/relayClient"; import { activateRateLimit } from "@/shared/api/relayRateLimitGate"; import { resolveAgentParallelism } from "@/features/agents/lib/agentParallelism"; @@ -13376,6 +13377,23 @@ export function maybeInstallE2eTauriMocks() { case "start_archive_sync": case "stop_archive_sync": return null; + case "unread_catch_up": { + const request = payload as { + request: { channels: Array<{ id: string }> }; + }; + const results: UnreadCatchUpChannelResult[] = + request.request.channels.map((channel) => ({ + status: "success", + channelId: channel.id, + observedEvents: [], + maxTrigger: 0, + activityRows: [], + discovered: { participated: [], authored: [], mentioned: [] }, + })); + // Keep this mock aligned with the complete Rust serde shape pinned by + // `serialized_response_matches_the_typescript_contract`. + return { channels: results }; + } case "agent_metric_archive_default_enabled": return activeConfig?.mock?.agentMetricArchiveDefaultEnabled ?? true; case "set_prevent_sleep_active":