From e1d48039f3567336382377b24e124db27eadf14e Mon Sep 17 00:00:00 2001 From: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 11:36:41 -0400 Subject: [PATCH 01/13] perf(desktop): persist channel head cache Co-authored-by: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz> Signed-off-by: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz> --- desktop/src-tauri/src/channel_head_cache.rs | 398 ++++++++++++++++++++ desktop/src-tauri/src/lib.rs | 5 + desktop/src-tauri/src/shutdown.rs | 1 + 3 files changed, 404 insertions(+) create mode 100644 desktop/src-tauri/src/channel_head_cache.rs diff --git a/desktop/src-tauri/src/channel_head_cache.rs b/desktop/src-tauri/src/channel_head_cache.rs new file mode 100644 index 00000000000..2cf2b7ef5ba --- /dev/null +++ b/desktop/src-tauri/src/channel_head_cache.rs @@ -0,0 +1,398 @@ +//! Persistent native cache for recently visited channel head pages. +//! +//! The cache is a paint accelerator only: the renderer always replaces a +//! hydrated page with an authoritative relay response after subscribing. + +use std::{ + path::{Path, PathBuf}, + sync::{Arc, Mutex}, +}; + +use rusqlite::{params, Connection, OptionalExtension, Transaction}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tauri::{AppHandle, Manager, State}; + +const SCHEMA_VERSION: i64 = 1; +const CHANNELS_PER_SCOPE_CAP: i64 = 32; +const ROW_BYTES_CAP: usize = 1024 * 1024; + +/// Serializes cache mutations on the blocking pool. +#[derive(Default)] +pub(crate) struct ChannelHeadCacheStore { + write_lock: Arc>, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ChannelHeadScope { + pub(crate) pubkey: String, + pub(crate) relay_url: String, +} + +impl ChannelHeadScope { + fn key(&self) -> String { + format!( + "{}:{}", + self.pubkey.trim().to_ascii_lowercase(), + self.relay_url.trim().trim_end_matches('/') + ) + } +} + +#[derive(Clone, Debug, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ChannelHeadEntry { + channel_id: String, + events: Vec, + saved_at: i64, + last_visited_at: i64, +} + +fn db_path(app: &AppHandle) -> Result { + let dir = app + .path() + .app_data_dir() + .map_err(|error| format!("resolve channel-head cache data dir: {error}"))?; + std::fs::create_dir_all(&dir) + .map_err(|error| format!("create channel-head cache data dir: {error}"))?; + Ok(dir.join("channel-head-cache.db")) +} + +fn create_schema(conn: &Connection) -> Result<(), String> { + conn.execute_batch( + "CREATE TABLE schema_meta(version INTEGER NOT NULL); + INSERT INTO schema_meta(version) VALUES(1); + CREATE TABLE channel_head( + scope TEXT NOT NULL, + channel_id TEXT NOT NULL, + events_json TEXT NOT NULL, + row_count INTEGER NOT NULL, + saved_at INTEGER NOT NULL, + last_visited_at INTEGER NOT NULL, + PRIMARY KEY(scope, channel_id) + );", + ) + .map_err(|error| format!("initialize channel-head cache db: {error}")) +} + +fn open_db(path: &Path) -> Result { + let conn = + Connection::open(path).map_err(|error| format!("open channel-head cache db: {error}"))?; + conn.pragma_update(None, "busy_timeout", 5_000) + .map_err(|error| format!("configure channel-head cache db: {error}"))?; + conn.pragma_update(None, "journal_mode", "WAL") + .map_err(|error| format!("configure channel-head cache WAL: {error}"))?; + + let has_schema_meta: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='schema_meta')", + [], + |row| row.get(0), + ) + .map_err(|error| format!("inspect channel-head cache schema: {error}"))?; + if !has_schema_meta { + create_schema(&conn)?; + return Ok(conn); + } + + let version = conn + .query_row("SELECT version FROM schema_meta LIMIT 1", [], |row| { + row.get::<_, i64>(0) + }) + .optional() + .map_err(|error| format!("read channel-head cache schema: {error}"))?; + let has_channel_head: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='channel_head')", + [], + |row| row.get(0), + ) + .map_err(|error| format!("inspect channel-head cache table: {error}"))?; + if version != Some(SCHEMA_VERSION) || !has_channel_head { + conn.execute_batch("DROP TABLE IF EXISTS channel_head; DROP TABLE IF EXISTS schema_meta;") + .map_err(|error| format!("reset channel-head cache schema: {error}"))?; + create_schema(&conn)?; + } + Ok(conn) +} + +async fn run_blocking(task: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + tauri::async_runtime::spawn_blocking(task) + .await + .map_err(|error| format!("channel-head cache db task failed: {error}"))? +} + +fn load_from_path( + path: &Path, + scope: &ChannelHeadScope, + limit: u32, +) -> Result, String> { + let conn = open_db(path)?; + let mut statement = conn + .prepare( + "SELECT channel_id, events_json, saved_at, last_visited_at + FROM channel_head WHERE scope=?1 + ORDER BY last_visited_at DESC, saved_at DESC, channel_id ASC LIMIT ?2", + ) + .map_err(|error| format!("prepare channel-head cache load: {error}"))?; + let rows = statement + .query_map(params![scope.key(), i64::from(limit)], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, i64>(3)?, + )) + }) + .map_err(|error| format!("query channel-head cache: {error}"))?; + let mut entries = Vec::new(); + for row in rows { + let (channel_id, events_json, saved_at, last_visited_at) = + row.map_err(|error| format!("read channel-head cache row: {error}"))?; + let events = serde_json::from_str(&events_json) + .map_err(|error| format!("decode channel-head cache row {channel_id}: {error}"))?; + entries.push(ChannelHeadEntry { + channel_id, + events, + saved_at, + last_visited_at, + }); + } + Ok(entries) +} + +fn store_in_transaction( + transaction: &Transaction<'_>, + scope: &str, + channel_id: &str, + events_json: &str, + row_count: usize, + now: i64, +) -> Result<(), String> { + let last_visited_at: i64 = transaction + .query_row( + "SELECT COALESCE(MAX(last_visited_at), ?2 - 1) + 1 FROM channel_head WHERE scope=?1", + params![scope, now], + |row| row.get(0), + ) + .map_err(|error| format!("advance channel-head cache visit clock: {error}"))?; + transaction + .execute( + "INSERT INTO channel_head(scope, channel_id, events_json, row_count, saved_at, last_visited_at) + VALUES(?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(scope, channel_id) DO UPDATE SET + events_json=excluded.events_json, + row_count=excluded.row_count, + saved_at=excluded.saved_at, + last_visited_at=excluded.last_visited_at", + params![scope, channel_id, events_json, row_count as i64, now, last_visited_at], + ) + .map_err(|error| format!("store channel-head cache row: {error}"))?; + transaction + .execute( + "DELETE FROM channel_head WHERE rowid IN ( + SELECT rowid FROM channel_head WHERE scope=?1 + ORDER BY last_visited_at DESC, saved_at DESC, channel_id ASC + LIMIT -1 OFFSET ?2 + )", + params![scope, CHANNELS_PER_SCOPE_CAP], + ) + .map_err(|error| format!("prune channel-head cache: {error}"))?; + Ok(()) +} + +fn store_at( + path: &Path, + scope: &ChannelHeadScope, + channel_id: &str, + events: &[Value], + now: i64, +) -> Result<(), String> { + let events_json = serde_json::to_string(events) + .map_err(|error| format!("encode channel-head cache row: {error}"))?; + let mut conn = open_db(path)?; + let transaction = conn + .transaction() + .map_err(|error| format!("begin channel-head cache store: {error}"))?; + if events_json.len() > ROW_BYTES_CAP { + transaction + .execute( + "DELETE FROM channel_head WHERE scope=?1 AND channel_id=?2", + params![scope.key(), channel_id], + ) + .map_err(|error| format!("drop oversized channel-head cache row: {error}"))?; + } else { + store_in_transaction( + &transaction, + &scope.key(), + channel_id, + &events_json, + events.len(), + now, + )?; + } + transaction + .commit() + .map_err(|error| format!("commit channel-head cache store: {error}")) +} + +/// Loads the most recently visited channel heads for one identity and relay. +#[tauri::command] +pub(crate) async fn channel_head_cache_load( + scope: ChannelHeadScope, + limit: u32, + app: AppHandle, +) -> Result, String> { + let path = db_path(&app)?; + run_blocking(move || load_from_path(&path, &scope, limit)).await +} + +/// Stores one raw channel-window response, dropping payloads above one MiB. +#[tauri::command] +pub(crate) async fn channel_head_cache_store( + scope: ChannelHeadScope, + channel_id: String, + events: Vec, + app: AppHandle, + store: State<'_, ChannelHeadCacheStore>, +) -> Result<(), String> { + let path = db_path(&app)?; + let write_lock = Arc::clone(&store.write_lock); + run_blocking(move || { + let _guard = write_lock.lock().map_err(|error| error.to_string())?; + store_at( + &path, + &scope, + &channel_id, + &events, + chrono::Utc::now().timestamp(), + ) + }) + .await +} + +/// Clears all persisted channel heads for one identity and relay. +#[tauri::command] +pub(crate) async fn channel_head_cache_clear( + scope: ChannelHeadScope, + app: AppHandle, + store: State<'_, ChannelHeadCacheStore>, +) -> Result<(), String> { + let path = db_path(&app)?; + let write_lock = Arc::clone(&store.write_lock); + run_blocking(move || { + let _guard = write_lock.lock().map_err(|error| error.to_string())?; + let conn = open_db(&path)?; + conn.execute("DELETE FROM channel_head WHERE scope=?1", [scope.key()]) + .map_err(|error| format!("clear channel-head cache scope: {error}"))?; + Ok(()) + }) + .await +} + +pub(crate) fn flush(app: &AppHandle) { + if let Ok(path) = db_path(app) { + if let Ok(conn) = open_db(&path) { + let _ = conn.execute_batch("PRAGMA wal_checkpoint(PASSIVE);"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scope() -> ChannelHeadScope { + ChannelHeadScope { + pubkey: "PK".into(), + relay_url: "wss://relay/".into(), + } + } + + #[test] + fn serialized_entry_matches_typescript_contract() { + let actual = serde_json::to_value(ChannelHeadEntry { + channel_id: "general".into(), + events: vec![serde_json::json!({"id":"event"})], + saved_at: 42, + last_visited_at: 43, + }) + .unwrap(); + let expected = serde_json::json!({ + "channelId":"general", + "events":[{"id":"event"}], + "savedAt":42, + "lastVisitedAt":43 + }); + assert_eq!(actual, expected); + + let decoded: ChannelHeadScope = serde_json::from_value(serde_json::json!({ + "pubkey":"PK", "relayUrl":"wss://relay/" + })) + .unwrap(); + assert_eq!(decoded, scope()); + assert_eq!(decoded.key(), "pk:wss://relay"); + } + + #[test] + fn enforces_lru_and_payload_caps() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("channel-head-cache.db"); + for index in 0..=CHANNELS_PER_SCOPE_CAP { + store_at( + &path, + &scope(), + &format!("channel-{index:02}"), + &[serde_json::json!({"index":index})], + 1_000 + index, + ) + .unwrap(); + } + let entries = load_from_path(&path, &scope(), 100).unwrap(); + assert_eq!(entries.len(), CHANNELS_PER_SCOPE_CAP as usize); + assert_eq!(entries.first().unwrap().channel_id, "channel-32"); + assert!(!entries.iter().any(|entry| entry.channel_id == "channel-00")); + + let oversized = vec![Value::String("x".repeat(ROW_BYTES_CAP))]; + store_at(&path, &scope(), "channel-32", &oversized, 2_000).unwrap(); + let count: i64 = open_db(&path) + .unwrap() + .query_row( + "SELECT COUNT(*) FROM channel_head WHERE channel_id='channel-32'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(count, 0); + } + + #[test] + fn schema_mismatch_recreates_cache() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("channel-head-cache.db"); + let conn = open_db(&path).unwrap(); + conn.execute("UPDATE schema_meta SET version=99", []) + .unwrap(); + conn.execute( + "INSERT INTO channel_head VALUES('scope','channel','[]',0,1,1)", + [], + ) + .unwrap(); + drop(conn); + + let reset = open_db(&path).unwrap(); + let version: i64 = reset + .query_row("SELECT version FROM schema_meta", [], |row| row.get(0)) + .unwrap(); + let rows: i64 = reset + .query_row("SELECT COUNT(*) FROM channel_head", [], |row| row.get(0)) + .unwrap(); + assert_eq!(version, SCHEMA_VERSION); + assert_eq!(rows, 0); + } +} diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 71a5eb3806e..428aa4d2a78 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -3,6 +3,7 @@ mod app_menu; mod app_state; mod archive; mod builderlab; +mod channel_head_cache; mod commands; mod deep_link; mod egress_guard; @@ -234,6 +235,7 @@ pub fn run() { .manage(archive::sync::ArchiveSyncState::default()) .manage(native_relay_client::NativeRelayClient::default()) .manage(observed_unread::ObservedUnreadStore::default()) + .manage(channel_head_cache::ChannelHeadCacheStore::default()) .setup(move |app| { let app_handle = app.handle().clone(); #[cfg(target_os = "macos")] @@ -724,6 +726,9 @@ pub fn run() { unread_catch_up::unread_catch_up, observed_unread::observed_unread_open_scope, observed_unread::observed_unread_ingest, + channel_head_cache::channel_head_cache_load, + channel_head_cache::channel_head_cache_store, + channel_head_cache::channel_head_cache_clear, list_personas, create_persona, update_persona, diff --git a/desktop/src-tauri/src/shutdown.rs b/desktop/src-tauri/src/shutdown.rs index 17ca7a7bb37..b1548c69370 100644 --- a/desktop/src-tauri/src/shutdown.rs +++ b/desktop/src-tauri/src/shutdown.rs @@ -20,6 +20,7 @@ pub(crate) fn shut_down_app(app: &tauri::AppHandle, shutdown_done: &std::sync::a if !shutdown_done.swap(true, Ordering::SeqCst) { prevent_sleep::release(&app.state::().prevent_sleep); crate::observed_unread::flush(app); + crate::channel_head_cache::flush(app); app.state::() .shutdown_all(); if let Err(error) = shutdown_managed_agents(app) { From f1e2e4dc9d7f60fdfa400376223fbc6a8c412ac8 Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 22 Aug 2026 11:49:14 -0400 Subject: [PATCH 02/13] perf(desktop): hydrate persisted channel heads Co-authored-by: Max Signed-off-by: Max --- desktop/src-tauri/src/app_state.rs | 4 +- desktop/src/app/App.tsx | 40 +++- .../src/app/useAppShellLifecycleEffects.ts | 28 +-- .../features/communities/useCommunities.tsx | 14 +- .../features/communities/useCommunityInit.ts | 8 +- desktop/src/features/messages/hooks.ts | 15 +- .../messages/lib/channelHeadCache.test.mjs | 99 +++++++++ .../features/messages/lib/channelHeadCache.ts | 76 +++++++ .../messages/lib/messageSnapshot.test.mjs | 199 ----------------- .../features/messages/lib/messageSnapshot.ts | 202 ------------------ desktop/src/features/profile/hooks.ts | 4 +- .../src/shared/api/tauriChannelHeadCache.ts | 25 +++ desktop/src/testing/e2eBridge.ts | 53 +++++ 13 files changed, 331 insertions(+), 436 deletions(-) create mode 100644 desktop/src/features/messages/lib/channelHeadCache.test.mjs create mode 100644 desktop/src/features/messages/lib/channelHeadCache.ts delete mode 100644 desktop/src/features/messages/lib/messageSnapshot.test.mjs delete mode 100644 desktop/src/features/messages/lib/messageSnapshot.ts create mode 100644 desktop/src/shared/api/tauriChannelHeadCache.ts diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 7c41f6bfe26..9cbb4444ab3 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -194,8 +194,8 @@ pub fn build_app_state() -> AppState { identity_storage: AtomicU8::new(identity_storage as u8), http_client: reqwest::Client::builder() .resolve("localhost", std::net::SocketAddr::from(([127, 0, 0, 1], 0))) - .pool_idle_timeout(std::time::Duration::from_secs(10)) - .pool_max_idle_per_host(1) + .pool_idle_timeout(std::time::Duration::from_secs(300)) + .pool_max_idle_per_host(2) .build() .unwrap_or_else(|_| reqwest::Client::new()), media_fetch_client: build_media_fetch_client().expect( diff --git a/desktop/src/app/App.tsx b/desktop/src/app/App.tsx index bfaf2ba2008..23899cfe145 100644 --- a/desktop/src/app/App.tsx +++ b/desktop/src/app/App.tsx @@ -63,6 +63,7 @@ import { CommunityChangeOverlay } from "@/features/communities/ui/CommunityChang import { setAvatarProfileSyncQueryClient } from "@/features/profile/avatarProfileSync"; import { EncryptedBackupProvider } from "@/features/settings/EncryptedBackupProvider"; import { createBuzzQueryClient } from "@/shared/api/queryClient"; +import { hydrateChannelHeads } from "@/features/messages/lib/channelHeadCache"; import { useIdentityQuery } from "@/shared/api/hooks"; import { isSharedIdentity as isSharedIdentityCmd } from "@/shared/api/tauri"; import { getProfile } from "@/shared/api/tauriProfiles"; @@ -213,11 +214,38 @@ function CommunitySwitchGate() { ); } -function CommunityQueryProvider({ children }: { children: ReactNode }) { +function CommunityQueryProvider({ + children, + pubkey, + relayUrl, +}: { + children: ReactNode; + pubkey: string | null; + relayUrl: string | null; +}) { const [queryClient] = useState(createBuzzQueryClient); + const [isHydrated, setIsHydrated] = useState(!pubkey || !relayUrl); useEffect(() => setAvatarProfileSyncQueryClient(queryClient), [queryClient]); + useEffect(() => { + let cancelled = false; + if (!pubkey || !relayUrl) { + setIsHydrated(true); + return; + } + void hydrateChannelHeads(queryClient, { pubkey, relayUrl }) + .catch((error) => { + console.warn("Failed to hydrate persisted channel heads", error); + }) + .finally(() => { + if (!cancelled) setIsHydrated(true); + }); + return () => { + cancelled = true; + }; + }, [pubkey, queryClient, relayUrl]); + useEffect(() => { const e2eWindow = window as Window & { __BUZZ_E2E__?: unknown; @@ -236,7 +264,9 @@ function CommunityQueryProvider({ children }: { children: ReactNode }) { }, [queryClient]); return ( - {children} + + {isHydrated ? children : null} + ); } @@ -601,7 +631,11 @@ function CommunityApp({ }, [communityApplied]); if (appContent === null && (!transaction || isEnteringCurtain)) { appContent = communityApplied ? ( - + diff --git a/desktop/src/app/useAppShellLifecycleEffects.ts b/desktop/src/app/useAppShellLifecycleEffects.ts index 969bf67ca67..71db0523695 100644 --- a/desktop/src/app/useAppShellLifecycleEffects.ts +++ b/desktop/src/app/useAppShellLifecycleEffects.ts @@ -42,33 +42,13 @@ export function useAppShellLifecycleEffects({ React.useEffect(() => { let isCancelled = false; - - const startPreconnect = () => { - if (isCancelled) { - return; + void relayClient.preconnect().catch((error) => { + if (!isCancelled) { + console.error("Failed to preconnect to relay", error); } - - void relayClient.preconnect().catch((error) => { - if (!isCancelled) { - console.error("Failed to preconnect to relay", error); - } - }); - }; - - if ("requestIdleCallback" in window) { - const idleId = window.requestIdleCallback(startPreconnect, { - timeout: 1_500, - }); - return () => { - isCancelled = true; - window.cancelIdleCallback(idleId); - }; - } - - const timeoutId = globalThis.setTimeout(startPreconnect, 250); + }); return () => { isCancelled = true; - globalThis.clearTimeout(timeoutId); }; }, []); diff --git a/desktop/src/features/communities/useCommunities.tsx b/desktop/src/features/communities/useCommunities.tsx index e0a10017883..f30353af72d 100644 --- a/desktop/src/features/communities/useCommunities.tsx +++ b/desktop/src/features/communities/useCommunities.tsx @@ -19,7 +19,8 @@ import { import { removeSelfProfileCachesForRelay } from "@/features/profile/lib/selfProfileStorage"; import { removeUserLabelCacheForRelay } from "@/features/profile/lib/userLabelStorage"; import { removeChannelSnapshotForRelay } from "@/features/channels/channelSnapshot"; -import { removeMessageSnapshotsForRelay } from "@/features/messages/lib/messageSnapshot"; +import { clearChannelHeadCache } from "@/shared/api/tauriChannelHeadCache"; +import { getIdentity } from "@/shared/api/tauriIdentity"; import { clearSavedCommunitySnapshot } from "@/features/agents/activeAgentTurnsStore"; import { clearCommunityDestinations, @@ -234,7 +235,16 @@ function useCommunitiesInternal(): UseCommunitiesReturn { removeSelfProfileCachesForRelay(removed.relayUrl); removeUserLabelCacheForRelay(removed.relayUrl); removeChannelSnapshotForRelay(removed.relayUrl); - removeMessageSnapshotsForRelay(removed.relayUrl); + void getIdentity() + .then((identity) => + clearChannelHeadCache({ + pubkey: identity.pubkey, + relayUrl: removed.relayUrl, + }), + ) + .catch((error) => { + console.warn("Failed to clear persisted channel heads", error); + }); clearSavedCommunitySnapshot(id); removeCommunityDestination(id); diff --git a/desktop/src/features/communities/useCommunityInit.ts b/desktop/src/features/communities/useCommunityInit.ts index 5493a47b1e3..c565ee0f7b4 100644 --- a/desktop/src/features/communities/useCommunityInit.ts +++ b/desktop/src/features/communities/useCommunityInit.ts @@ -84,7 +84,12 @@ async function resetCommunityState({ } type CommunityInitResult = - | { isReady: true; needsSetup: false; appliedKey: string } + | { + isReady: true; + needsSetup: false; + appliedKey: string; + identityPubkey: string | null; + } | { isReady: false; needsSetup: true; @@ -342,6 +347,7 @@ export function useCommunityInit( isReady: true, needsSetup: false, appliedKey: communityKey, + identityPubkey, }); } } diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 8b457a7adf8..ff022ab522b 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -24,6 +24,11 @@ import { refreshChannelWindowMessages, } from "@/features/messages/lib/projectChannelWindow"; import { reconcileChannelWindowMessages } from "@/features/messages/lib/channelWindowReconciliation"; +import { + channelHeadCacheScope, + consumeHydratedChannel, +} from "@/features/messages/lib/channelHeadCache"; +import { storeChannelHeadCache } from "@/shared/api/tauriChannelHeadCache"; import { mergeMessages, mergeTimelineCacheMessages, @@ -257,18 +262,26 @@ export function reconcileFetchedChannelWindow( emptyChannelWindowStore(); const next = replaceNewestChannelWindow(current, page); queryClient.setQueryData(windowKey, next); + const scope = channelHeadCacheScope(queryClient); + if (scope) { + void storeChannelHeadCache(scope, channelId, events).catch((error) => { + console.warn("Failed to persist channel head", channelId, error); + }); + } return reconcileChannelWindowMessages(next, previousMessages); } export function useChannelMessagesQuery(channel: Channel | null) { const queryClient = useQueryClient(); const queryKey = channelMessagesKey(channel?.id ?? "none"); - return useQuery({ enabled: channel !== null && channel.channelType !== "forum", queryKey, queryFn: async ({ signal }) => { if (!channel) throw new Error("No channel selected."); + if (consumeHydratedChannel(queryClient, channel.id)) { + return queryClient.getQueryData(queryKey) ?? []; + } const previousMessages = queryClient.getQueryData(queryKey) ?? []; const events = await getChannelWindowEvents(channel.id); diff --git a/desktop/src/features/messages/lib/channelHeadCache.test.mjs b/desktop/src/features/messages/lib/channelHeadCache.test.mjs new file mode 100644 index 00000000000..31d7076d5ef --- /dev/null +++ b/desktop/src/features/messages/lib/channelHeadCache.test.mjs @@ -0,0 +1,99 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { QueryClient } from "@tanstack/react-query"; +import { + consumeHydratedChannel, + hydrateChannelHeads, +} from "./channelHeadCache.ts"; +import { channelMessagesKey } from "./messageQueryKeys.ts"; +import { reconcileFetchedChannelWindow } from "../hooks.ts"; +const channelId = "channel-a"; +const root = { + id: "a".repeat(64), + pubkey: "b".repeat(64), + created_at: 10, + kind: 40002, + tags: [["h", channelId]], + content: "persisted", + sig: "", +}; +const replacement = { + ...root, + id: "c".repeat(64), + created_at: 11, + content: "relay", +}; +function bounds() { + return { + id: "d".repeat(64), + pubkey: "e".repeat(64), + created_at: 12, + kind: 39006, + tags: [ + ["h", channelId], + ["d", `${channelId}:head`], + ], + content: JSON.stringify({ has_more: false, next_cursor: null }), + sig: "", + }; +} +function install(entries) { + globalThis.window = { + localStorage: { getItem: () => null }, + __TAURI_INTERNALS__: { + invoke: async (command) => + command === "channel_head_cache_load" ? entries : null, + }, + }; +} +test("hydrates stale data and consumes its mount gate once", async () => { + install([ + { channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }, + ]); + const client = new QueryClient(); + await hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + assert.deepEqual(client.getQueryData(channelMessagesKey(channelId)), [root]); + assert.equal( + client.getQueryState(channelMessagesKey(channelId)).dataUpdatedAt, + 0, + ); + assert.equal(consumeHydratedChannel(client, channelId), true); + assert.equal(consumeHydratedChannel(client, channelId), false); +}); +test("authoritative refresh deletes a vanished hydrated row", async () => { + install([ + { channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }, + ]); + const client = new QueryClient(); + await hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + const next = reconcileFetchedChannelWindow( + client, + channelId, + [replacement, bounds()], + client.getQueryData(channelMessagesKey(channelId)), + new AbortController().signal, + ); + assert.deepEqual( + next.map((e) => e.id), + [replacement.id], + ); +}); +test("drops malformed entries independently", async () => { + install([ + { channelId: "bad", events: [root], savedAt: 1, lastVisitedAt: 2 }, + { channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }, + ]); + const client = new QueryClient(); + await hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + assert.equal(client.getQueryData(channelMessagesKey("bad")), undefined); + assert.deepEqual(client.getQueryData(channelMessagesKey(channelId)), [root]); +}); diff --git a/desktop/src/features/messages/lib/channelHeadCache.ts b/desktop/src/features/messages/lib/channelHeadCache.ts new file mode 100644 index 00000000000..409543be181 --- /dev/null +++ b/desktop/src/features/messages/lib/channelHeadCache.ts @@ -0,0 +1,76 @@ +import type { QueryClient } from "@tanstack/react-query"; +import { + loadChannelHeadCache, + type ChannelHeadScope, +} from "@/shared/api/tauriChannelHeadCache"; +import { channelMessagesKey, channelWindowKey } from "./messageQueryKeys"; +import { parseChannelWindowResponse } from "./channelWindowResponse"; +import { + emptyChannelWindowStore, + replaceNewestChannelWindow, +} from "./channelWindowStore"; +import { reconcileChannelWindowMessages } from "./channelWindowReconciliation"; +const hydratedChannels = new WeakMap>(); +const cacheScopes = new WeakMap(); +export function isChannelHeadCacheEnabled(): boolean { + if (typeof window === "undefined") return false; + if (import.meta.env?.VITE_BUZZ_CHANNEL_HEAD_CACHE === "off") return false; + return window.localStorage.getItem("buzz-channel-head-cache") !== "off"; +} +export function channelHeadCacheScope( + queryClient: QueryClient, +): ChannelHeadScope | null { + return cacheScopes.get(queryClient) ?? null; +} +export function consumeHydratedChannel( + queryClient: QueryClient, + channelId: string, +): boolean { + const channels = hydratedChannels.get(queryClient); + if (!channels?.delete(channelId)) return false; + if (channels.size === 0) hydratedChannels.delete(queryClient); + return true; +} +export function hasHydratedChannel( + queryClient: QueryClient, + channelId: string, +): boolean { + return hydratedChannels.get(queryClient)?.has(channelId) ?? false; +} +export async function hydrateChannelHeads( + queryClient: QueryClient, + scope: ChannelHeadScope, +): Promise { + if (!isChannelHeadCacheEnabled()) return; + cacheScopes.set(queryClient, scope); + const entries = await loadChannelHeadCache(scope, 12); + const hydrated = new Set(); + for (const entry of entries) { + try { + const page = parseChannelWindowResponse( + entry.events, + entry.channelId, + null, + ); + const window = replaceNewestChannelWindow( + emptyChannelWindowStore(), + page, + ); + const messages = reconcileChannelWindowMessages(window, []); + queryClient.setQueryData(channelWindowKey(entry.channelId), window, { + updatedAt: 0, + }); + queryClient.setQueryData(channelMessagesKey(entry.channelId), messages, { + updatedAt: 0, + }); + hydrated.add(entry.channelId); + } catch (error) { + console.warn( + "Ignoring invalid persisted channel head", + entry.channelId, + error, + ); + } + } + if (hydrated.size > 0) hydratedChannels.set(queryClient, hydrated); +} diff --git a/desktop/src/features/messages/lib/messageSnapshot.test.mjs b/desktop/src/features/messages/lib/messageSnapshot.test.mjs deleted file mode 100644 index a188f51da08..00000000000 --- a/desktop/src/features/messages/lib/messageSnapshot.test.mjs +++ /dev/null @@ -1,199 +0,0 @@ -import assert from "node:assert/strict"; -import test from "node:test"; - -import { - mergeHistoryOverSnapshot, - messageSnapshotKey, - readMessageSnapshot, - removeMessageSnapshotsForRelay, - writeMessageSnapshot, -} from "./messageSnapshot.ts"; - -if (typeof globalThis.window === "undefined") { - const storage = new Map(); - globalThis.window = { - localStorage: { - getItem: (key) => storage.get(key) ?? null, - setItem: (key, value) => storage.set(key, value), - removeItem: (key) => storage.delete(key), - key: (index) => [...storage.keys()][index] ?? null, - get length() { - return storage.size; - }, - }, - }; -} - -function makeEvent(overrides = {}) { - return { - id: `event-${Math.random().toString(36).slice(2)}`, - pubkey: "pubkey-1", - created_at: 1_700_000_000, - kind: 9, - tags: [["h", "chan-1"]], - content: "hello", - sig: "sig", - ...overrides, - }; -} - -const RELAY = "wss://relay.example.com"; - -function clearRelay(relayUrl = RELAY) { - removeMessageSnapshotsForRelay(relayUrl); -} - -test("messageSnapshotKey: normalizes trailing slash and case", () => { - assert.equal( - messageSnapshotKey("WSS://Relay.Example.com/", "chan-1"), - messageSnapshotKey("wss://relay.example.com", "chan-1"), - ); -}); - -test("read after write returns the persisted events", () => { - clearRelay(); - const events = [makeEvent({ id: "a" }), makeEvent({ id: "b" })]; - writeMessageSnapshot(RELAY, "chan-1", events); - assert.deepEqual(readMessageSnapshot(RELAY, "chan-1"), events); -}); - -test("read for an unknown channel returns null", () => { - clearRelay(); - assert.equal(readMessageSnapshot(RELAY, "chan-never"), null); -}); - -test("read returns null for malformed JSON", () => { - window.localStorage.setItem( - messageSnapshotKey(RELAY, "chan-bad"), - "not-json{{{", - ); - assert.equal(readMessageSnapshot(RELAY, "chan-bad"), null); -}); - -test("read returns null for a wrong-version payload", () => { - window.localStorage.setItem( - messageSnapshotKey(RELAY, "chan-v2"), - JSON.stringify({ version: 2, updatedAt: 1, events: [makeEvent()] }), - ); - assert.equal(readMessageSnapshot(RELAY, "chan-v2"), null); -}); - -test("pending optimistic events are not persisted", () => { - clearRelay(); - const settled = makeEvent({ id: "settled" }); - writeMessageSnapshot(RELAY, "chan-1", [ - settled, - makeEvent({ id: "optimistic", pending: true }), - ]); - assert.deepEqual(readMessageSnapshot(RELAY, "chan-1"), [settled]); -}); - -test("write with only pending events persists nothing", () => { - clearRelay(); - writeMessageSnapshot(RELAY, "chan-1", [makeEvent({ pending: true })]); - assert.equal(readMessageSnapshot(RELAY, "chan-1"), null); -}); - -test("snapshot keeps only the newest slice of a long timeline", () => { - clearRelay(); - const events = Array.from({ length: 200 }, (_, i) => - makeEvent({ id: `event-${i}`, created_at: 1_700_000_000 + i }), - ); - writeMessageSnapshot(RELAY, "chan-1", events); - const persisted = readMessageSnapshot(RELAY, "chan-1"); - assert.equal(persisted.length, 80); - assert.equal(persisted[persisted.length - 1].id, "event-199"); - assert.equal(persisted[0].id, "event-120"); -}); - -test("per-relay channel cap evicts the least recently written snapshot", () => { - clearRelay(); - for (let i = 0; i < 21; i++) { - writeMessageSnapshot(RELAY, `chan-${i}`, [makeEvent({ id: `e-${i}` })]); - } - // chan-0 was written first (oldest updatedAt tie broken by insertion) — - // with 21 channels, at least one of the earliest must be evicted and the - // newest retained. - assert.notEqual(readMessageSnapshot(RELAY, "chan-20"), null); - const retained = Array.from({ length: 21 }, (_, i) => - readMessageSnapshot(RELAY, `chan-${i}`), - ).filter((snapshot) => snapshot !== null); - assert.equal(retained.length, 20); -}); - -test("remove clears every snapshot for that relay only", () => { - clearRelay(); - clearRelay("wss://other.example.com"); - writeMessageSnapshot(RELAY, "chan-1", [makeEvent({ id: "keep-other" })]); - writeMessageSnapshot("wss://other.example.com", "chan-1", [ - makeEvent({ id: "other" }), - ]); - removeMessageSnapshotsForRelay(RELAY); - assert.equal(readMessageSnapshot(RELAY, "chan-1"), null); - assert.notEqual( - readMessageSnapshot("wss://other.example.com", "chan-1"), - null, - ); -}); - -test("write is tolerant of storage failures", () => { - const original = window.localStorage.setItem; - window.localStorage.setItem = () => { - throw new Error("quota exceeded"); - }; - try { - assert.doesNotThrow(() => - writeMessageSnapshot(RELAY, "chan-1", [makeEvent()]), - ); - } finally { - window.localStorage.setItem = original; - } -}); - -test("cold snapshot load: merge keeps snapshot-only rows and widens aux backfill to them", () => { - const snapshotOnly = makeEvent({ id: "ghost", created_at: 1_700_000_000 }); - const fresh = makeEvent({ id: "fresh", created_at: 1_700_000_100 }); - const { merged, auxBackfillWindow } = mergeHistoryOverSnapshot({ - cached: undefined, - snapshot: [snapshotOnly], - history: [fresh], - }); - assert.deepEqual( - merged.map((event) => event.id), - ["ghost", "fresh"], - ); - assert.ok(auxBackfillWindow.some((event) => event.id === "ghost")); - assert.ok(auxBackfillWindow.some((event) => event.id === "fresh")); -}); - -test("warm load: aux backfill stays scoped to the fresh window", () => { - const cached = makeEvent({ id: "cached", created_at: 1_700_000_000 }); - const fresh = makeEvent({ id: "fresh", created_at: 1_700_000_100 }); - const { merged, auxBackfillWindow } = mergeHistoryOverSnapshot({ - cached: [cached], - snapshot: [makeEvent({ id: "stale-snapshot" })], - history: [fresh], - }); - assert.ok(merged.some((event) => event.id === "cached")); - assert.deepEqual( - auxBackfillWindow.map((event) => event.id), - ["fresh"], - ); -}); - -test("cold load without a snapshot backfills the fresh window only", () => { - const fresh = makeEvent({ id: "fresh" }); - const { merged, auxBackfillWindow } = mergeHistoryOverSnapshot({ - cached: undefined, - snapshot: null, - history: [fresh], - }); - assert.deepEqual( - merged.map((event) => event.id), - ["fresh"], - ); - assert.deepEqual( - auxBackfillWindow.map((event) => event.id), - ["fresh"], - ); -}); diff --git a/desktop/src/features/messages/lib/messageSnapshot.ts b/desktop/src/features/messages/lib/messageSnapshot.ts deleted file mode 100644 index d4a183c2ff6..00000000000 --- a/desktop/src/features/messages/lib/messageSnapshot.ts +++ /dev/null @@ -1,202 +0,0 @@ -/** - * Per-channel persisted message snapshots. - * - * A channel revisited after its React-Query cache entry is gone (app restart, - * gcTime expiry, community remount) goes fully cold and holds a skeleton for a - * relay round trip. This module persists the newest slice of each channel's - * timeline so a revisit can paint instantly from the snapshot while the - * history fetch revalidates behind it — the same stale-then-revalidate pattern - * the sidebar's channelSnapshot uses for the channel list. - * - * Keyed per relay URL + channel id so one relay's messages never bleed into - * another. Bounded two ways: only the newest MAX_EVENTS_PER_SNAPSHOT events - * per channel, and only the MAX_CHANNELS_PER_RELAY most recently written - * channels per relay (older ones are evicted LRU on write). - */ - -import { mergeTimelineHistoryMessages } from "@/features/messages/lib/messageQueryKeys"; -import { normalizeRelayUrl } from "@/features/profile/lib/selfProfileStorage"; -import type { RelayEvent } from "@/shared/api/types"; -import { setLocalStorageItemWithRecovery } from "@/shared/lib/localStorageQuota"; - -const STORAGE_KEY_PREFIX = "buzz-channel-messages.v1"; - -// Newest events kept per channel. The trailing slice of the sorted timeline -// cache, so recent auxiliary events (reactions/edits) ride along with the -// content rows they decorate. -const MAX_EVENTS_PER_SNAPSHOT = 80; - -const MAX_CHANNELS_PER_RELAY = 20; - -export function messageSnapshotKey(relayUrl: string, channelId: string) { - return `${STORAGE_KEY_PREFIX}:${normalizeRelayUrl(relayUrl)}:${channelId}`; -} - -type SnapshotPayload = { - version: 1; - updatedAt: number; - events: RelayEvent[]; -}; - -function parseSnapshotPayload(json: unknown): SnapshotPayload | null { - if (typeof json !== "object" || json === null) return null; - const obj = json as Record; - if (obj.version !== 1 || !Array.isArray(obj.events)) return null; - const updatedAt = - typeof obj.updatedAt === "number" && Number.isFinite(obj.updatedAt) - ? obj.updatedAt - : 0; - return { version: 1, updatedAt, events: obj.events as RelayEvent[] }; -} - -/** - * Reads the persisted message snapshot for a channel, or null when absent or - * malformed. - */ -export function readMessageSnapshot( - relayUrl: string, - channelId: string, -): RelayEvent[] | null { - try { - const raw = window.localStorage.getItem( - messageSnapshotKey(relayUrl, channelId), - ); - if (!raw) return null; - const parsed = parseSnapshotPayload(JSON.parse(raw)); - if (!parsed || parsed.events.length === 0) return null; - return parsed.events; - } catch { - return null; - } -} - -function relayPrefix(relayUrl: string) { - return `${STORAGE_KEY_PREFIX}:${normalizeRelayUrl(relayUrl)}:`; -} - -function collectKeysWithPrefix(prefix: string): string[] { - const keys: string[] = []; - for (let i = 0; i < window.localStorage.length; i++) { - const key = window.localStorage.key(i); - if (key?.startsWith(prefix)) { - keys.push(key); - } - } - return keys; -} - -function evictOldestSnapshots(prefix: string, keepingKey: string) { - const others = collectKeysWithPrefix(prefix).filter( - (key) => key !== keepingKey, - ); - if (others.length < MAX_CHANNELS_PER_RELAY) { - return; - } - - const byAge = others - .map((key) => { - let updatedAt = 0; - try { - const parsed = parseSnapshotPayload( - JSON.parse(window.localStorage.getItem(key) ?? ""), - ); - updatedAt = parsed?.updatedAt ?? 0; - } catch { - // Malformed entries sort oldest and get evicted first. - } - return { key, updatedAt }; - }) - .sort((a, b) => a.updatedAt - b.updatedAt); - - for (const { key } of byAge.slice( - 0, - others.length - (MAX_CHANNELS_PER_RELAY - 1), - )) { - window.localStorage.removeItem(key); - } -} - -/** - * Persists the newest slice of a channel's timeline. Pending optimistic events - * are dropped (they have no relay identity to revalidate against). Skips the - * write when unchanged so live-append churn does not re-serialize an identical - * snapshot. Non-fatal on storage failure (e.g. quota exceeded). - */ -export function writeMessageSnapshot( - relayUrl: string, - channelId: string, - events: RelayEvent[], -): void { - try { - const persistable = events - .filter((event) => !event.pending) - .slice(-MAX_EVENTS_PER_SNAPSHOT); - if (persistable.length === 0) { - return; - } - - const key = messageSnapshotKey(relayUrl, channelId); - const previous = window.localStorage.getItem(key); - if (previous) { - const parsed = parseSnapshotPayload(JSON.parse(previous)); - if ( - parsed && - JSON.stringify(parsed.events) === JSON.stringify(persistable) - ) { - return; - } - } - - evictOldestSnapshots(relayPrefix(relayUrl), key); - setLocalStorageItemWithRecovery( - key, - JSON.stringify({ - version: 1, - updatedAt: Date.now(), - events: persistable, - } satisfies SnapshotPayload), - ); - } catch { - // Storage access failures are non-fatal. - } -} - -/** - * Merge a fresh history window over the in-memory cache — or, when cold, over - * the persisted snapshot — and pick the window aux backfill must cover. - * - * The snapshot can hold events older than the fetch window; dropping them on - * settle would visibly shrink an already-painted timeline, so the merge keeps - * them. But a kept snapshot row deleted/edited while the app was closed never - * reappears in any history fetch (the relay soft-deletes), so its tombstone or - * edit is only reachable by `#e` over that row's id — cold snapshot loads must - * therefore backfill over the merged timeline, not just the fresh window. - * Otherwise the ghost paints, and the post-settle snapshot rewrite persists it - * forever. - */ -export function mergeHistoryOverSnapshot(input: { - cached: RelayEvent[] | undefined; - snapshot: RelayEvent[] | null; - history: RelayEvent[]; -}): { merged: RelayEvent[]; auxBackfillWindow: RelayEvent[] } { - const usedSnapshot = !input.cached && input.snapshot !== null; - const merged = mergeTimelineHistoryMessages( - input.cached ?? input.snapshot ?? [], - input.history, - ); - return { merged, auxBackfillWindow: usedSnapshot ? merged : input.history }; -} - -/** - * Removes every channel message snapshot for a relay. Called when a community - * is removed. - */ -export function removeMessageSnapshotsForRelay(relayUrl: string): void { - try { - for (const key of collectKeysWithPrefix(relayPrefix(relayUrl))) { - window.localStorage.removeItem(key); - } - } catch { - // Storage access failures are non-fatal. - } -} diff --git a/desktop/src/features/profile/hooks.ts b/desktop/src/features/profile/hooks.ts index f174d504080..233b1286125 100644 --- a/desktop/src/features/profile/hooks.ts +++ b/desktop/src/features/profile/hooks.ts @@ -351,7 +351,7 @@ export function useUsersBatchQuery( const entry = queryClient.getQueryData( usersBatchEntryKey(pubkey), ); - if (entry && now - entry.fetchedAt < 60_000) { + if (entry && now - entry.fetchedAt < 10 * 60_000) { if (entry.summary) profiles[pubkey] = entry.summary; else missing.push(pubkey); } else { @@ -384,7 +384,7 @@ export function useUsersBatchQuery( relayUrl, normalizedPubkeys, ), - staleTime: 60_000, + staleTime: 10 * 60_000, gcTime: 5 * 60 * 1_000, }); diff --git a/desktop/src/shared/api/tauriChannelHeadCache.ts b/desktop/src/shared/api/tauriChannelHeadCache.ts new file mode 100644 index 00000000000..9431fced046 --- /dev/null +++ b/desktop/src/shared/api/tauriChannelHeadCache.ts @@ -0,0 +1,25 @@ +import { invokeTauri } from "@/shared/api/tauri"; +import type { RelayEvent } from "@/shared/api/types"; +export type ChannelHeadScope = { pubkey: string; relayUrl: string }; +export type ChannelHeadEntry = { + channelId: string; + events: RelayEvent[]; + savedAt: number; + lastVisitedAt: number; +}; +export function loadChannelHeadCache( + scope: ChannelHeadScope, + limit = 12, +): Promise { + return invokeTauri("channel_head_cache_load", { scope, limit }); +} +export function storeChannelHeadCache( + scope: ChannelHeadScope, + channelId: string, + events: RelayEvent[], +): Promise { + return invokeTauri("channel_head_cache_store", { scope, channelId, events }); +} +export function clearChannelHeadCache(scope: ChannelHeadScope): Promise { + return invokeTauri("channel_head_cache_clear", { scope }); +} diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index e4028f01716..63f3a4ba385 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -13889,6 +13889,59 @@ export function maybeInstallE2eTauriMocks() { return null; case "fetch_persona_catalog": return mockPersonaCatalogPublications(); + case "channel_head_cache_load": { + const args = payload as { + scope: { pubkey: string; relayUrl: string }; + limit: number; + }; + const key = `buzz-e2e-channel-head:${args.scope.pubkey.toLowerCase()}:${args.scope.relayUrl.toLowerCase().replace(/\/$/, "")}`; + const entries = JSON.parse( + window.localStorage.getItem(key) ?? "[]", + ) as Array<{ + channelId: string; + events: RelayEvent[]; + savedAt: number; + lastVisitedAt: number; + }>; + return entries + .sort((left, right) => right.lastVisitedAt - left.lastVisitedAt) + .slice(0, args.limit); + } + case "channel_head_cache_store": { + const args = payload as { + scope: { pubkey: string; relayUrl: string }; + channelId: string; + events: RelayEvent[]; + }; + const key = `buzz-e2e-channel-head:${args.scope.pubkey.toLowerCase()}:${args.scope.relayUrl.toLowerCase().replace(/\/$/, "")}`; + const entries = JSON.parse( + window.localStorage.getItem(key) ?? "[]", + ) as Array<{ + channelId: string; + events: RelayEvent[]; + savedAt: number; + lastVisitedAt: number; + }>; + const now = Math.floor(Date.now() / 1000); + const next = entries + .filter((entry) => entry.channelId !== args.channelId) + .concat({ + channelId: args.channelId, + events: args.events, + savedAt: now, + lastVisitedAt: now, + }) + .sort((left, right) => right.lastVisitedAt - left.lastVisitedAt) + .slice(0, 32); + window.localStorage.setItem(key, JSON.stringify(next)); + return null; + } + case "channel_head_cache_clear": { + const args = payload as { scope: { pubkey: string; relayUrl: string } }; + const key = `buzz-e2e-channel-head:${args.scope.pubkey.toLowerCase()}:${args.scope.relayUrl.toLowerCase().replace(/\/$/, "")}`; + window.localStorage.removeItem(key); + return null; + } case "observed_unread_open_scope": { const request = payload as { request: { From 9d7623b3b56abfd2cf263fc47c3f20ba6fc2f0c2 Mon Sep 17 00:00:00 2001 From: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 11:40:53 -0400 Subject: [PATCH 03/13] perf(threads): collapse reply reads and sends Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> --- crates/buzz-relay/src/api/bridge.rs | 67 ++++++++++++++++-- desktop/src-tauri/src/commands/messages.rs | 68 +++++++++---------- .../src/commands/messages/thread_ref.rs | 34 +++++++++- .../src-tauri/src/commands/messages_tests.rs | 12 ++++ desktop/src/features/messages/hooks.ts | 28 +++++++- .../messages/lib/sendChannelBinding.test.mjs | 21 ++++++ .../messages/useThreadReplies.test.mjs | 35 +++------- .../src/features/messages/useThreadReplies.ts | 67 +----------------- desktop/src/shared/api/tauriMessages.ts | 2 + desktop/src/testing/e2eBridge.ts | 14 ++-- 10 files changed, 214 insertions(+), 134 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 8fdea4b3c02..ee4bd081624 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -393,6 +393,18 @@ const WINDOW_AUX_DELETE_KINDS: [u32; 2] = [ buzz_core::kind::KIND_NIP29_DELETE_EVENT, ]; +fn build_aux_query( + community: buzz_core::CommunityId, + target_ids: Vec, + kinds: &[u32], +) -> buzz_db::EventQuery { + let mut query = buzz_db::EventQuery::for_community(community); + query.kinds = Some(kinds.iter().map(|kind| *kind as i32).collect()); + query.e_tags = Some(target_ids); + query.limit = Some(1000); + query +} + /// Serve one `top_level: true` channel-window filter on the bridge `/query` /// path (docs/bridge-channel-window.md). Appends, in order: row events, the /// aux closure (`include_aux`), `39005` thread-summary overlays @@ -496,10 +508,8 @@ async fn handle_channel_window_filter( std::collections::HashSet::new(); let mut hop_ids = row_ids_hex.clone(); for hop_kinds in [&WINDOW_AUX_KINDS[..], &WINDOW_AUX_DELETE_KINDS[..]] { - let mut aux_query = buzz_db::EventQuery::for_community(tenant.community()); - aux_query.kinds = Some(hop_kinds.iter().map(|k| *k as i32).collect()); - aux_query.e_tags = Some(std::mem::take(&mut hop_ids)); - aux_query.limit = Some(1000); + let aux_query = + build_aux_query(tenant.community(), std::mem::take(&mut hop_ids), hop_kinds); let aux_events = session .query_events(&aux_query) .await @@ -1203,6 +1213,8 @@ async fn query_events_authed( .await .map_err(|e| internal_error(&format!("thread query error: {e}")))?; + let mut thread_row_ids = Vec::with_capacity(thread_replies.len() + 1); + thread_row_ids.push(root_hex.to_string()); for reply in thread_replies { let se = reply.stored_event; if !event_in_accessible_channel(&se, &accessible_channels) { @@ -1214,10 +1226,43 @@ async fn query_events_authed( if !buzz_core::filter::reader_authorized_for_event(&se.event, &authed_pubkey_hex) { continue; } + thread_row_ids.push(se.event.id.to_hex()); if let Ok(v) = serde_json::to_value(&se.event) { events.push(v); } } + + if extension_flag(raw, "include_aux") && !thread_row_ids.is_empty() { + let mut seen_aux = std::collections::HashSet::new(); + let mut hop_ids = thread_row_ids; + for hop_kinds in [&WINDOW_AUX_KINDS[..], &WINDOW_AUX_DELETE_KINDS[..]] { + let aux_query = + build_aux_query(tenant.community(), std::mem::take(&mut hop_ids), hop_kinds); + let aux_events = state + .db + .query_events_routed("bridge_thread_aux", &aux_query) + .await + .map_err(|e| internal_error(&format!("thread aux query error: {e}")))?; + for se in aux_events { + if !seen_aux.insert(se.event.id) + || !event_in_accessible_channel(&se, &accessible_channels) + || !buzz_core::filter::reader_authorized_for_event( + &se.event, + &authed_pubkey_hex, + ) + { + continue; + } + hop_ids.push(se.event.id.to_hex()); + if let Ok(value) = serde_json::to_value(&se.event) { + events.push(value); + } + } + if hop_ids.is_empty() { + break; + } + } + } handled.insert(idx); } @@ -2373,6 +2418,20 @@ mod tests { assert!(!has_mixed_search_filters(&filters)); } + #[test] + fn thread_aux_query_targets_root_and_replies_with_full_first_hop() { + let tenant = fresh_tenant("relay.example"); + let targets = vec!["root".to_string(), "reply".to_string()]; + let query = build_aux_query(tenant.community(), targets.clone(), &WINDOW_AUX_KINDS); + + assert_eq!(query.e_tags, Some(targets)); + assert_eq!( + query.kinds, + Some(WINDOW_AUX_KINDS.iter().map(|kind| *kind as i32).collect()) + ); + assert_eq!(query.limit, Some(1000)); + } + #[test] fn bridge_search_mode_extension_defaults_to_full_text() { assert_eq!( diff --git a/desktop/src-tauri/src/commands/messages.rs b/desktop/src-tauri/src/commands/messages.rs index 31559777d2b..461f29e7fa6 100644 --- a/desktop/src-tauri/src/commands/messages.rs +++ b/desktop/src-tauri/src/commands/messages.rs @@ -236,22 +236,11 @@ fn search_messages_limit(limit: Option) -> u32 { limit.unwrap_or(20).min(500) } -/// Fetch the full reply subtree under a thread root, server-side. -/// -/// Unlike the channel timeline (which the desktop assembles from its local -/// cache by grouping on `e`-root tags), this walks `thread_metadata` on the -/// relay via `get_thread_replies`, so a thread renders complete even when its -/// replies fell outside the channel cold-load window. Results are chronological -/// (oldest first) and are the *replies* under the root (depth >= 1); the root -/// event itself is NOT returned (the relay query keys on `root_event_id`, and a -/// root row has no `root_event_id`). Callers already hold the root — it is the -/// open thread head — so this closes the descendant gap without re-fetching it. +/// Fetch the reply subtree and its auxiliary events under a thread root. /// /// Paging is forward keyset on `(created_at, event_id)`: pass the `next_cursor` /// from a previous page back as `cursor` to fetch the next batch. The event-id -/// tiebreak is required because replies routinely share a `created_at` second; -/// a timestamp-only cursor would skip every tied reply past the page limit. -/// `next_cursor` is `Some` only when a full page was returned. +/// tiebreak prevents same-second replies from being skipped. #[tauri::command] pub async fn get_thread_replies( root_event_id: String, @@ -275,8 +264,12 @@ pub async fn get_thread_replies( // A full page implies there may be more; hand back the last event's // composite key as the next cursor (the DB returns replies strictly after // it, tiebroken by event_id so same-second replies are not skipped). - let next_cursor = if events.len() as u32 >= cap { - events.last().map(|ev| crate::models::ThreadCursor { + let reply_events: Vec<_> = events + .iter() + .filter(|event| TIMELINE_KINDS.contains(&(event.kind.as_u16() as u32))) + .collect(); + let next_cursor = if reply_events.len() as u32 >= cap { + reply_events.last().map(|ev| crate::models::ThreadCursor { created_at: ev.created_at.as_secs() as i64, event_id: ev.id.to_hex(), }) @@ -295,21 +288,9 @@ pub async fn get_thread_replies( }) } -/// Build the relay `/query` filter for the server-side thread-subtree read. -/// -/// The relay routes a filter to `get_thread_replies` purely off a single `#e` -/// (root) tag plus `depth_limit` — kind is NOT part of that routing or the -/// underlying DB query (it keys on `root_event_id`). Yet `kinds` is still -/// required here: the bridge runs the p-gate (`p_gated_filters_authorized`) on -/// every filter *before* routing, and a kindless filter "could match" a p-gated -/// kind, so the gate demands a `#p` tag we don't send -> HTTP 403 -/// `restricted: p-gated kinds require #p tag`, before the thread query ever -/// runs. Carrying non-p-gated [`TIMELINE_KINDS`] makes the filter provably -/// un-p-gated so it clears the gate. `build_channel_messages_before_filter` is -/// the sibling that already does this, which is why the dense-second channel -/// pager was never gated and this reader was. Extracted so a unit test can pin -/// that `kinds` is present (the e2e mock does not model p-gating, so only a -/// unit test guards this contract). +/// Build the relay `/query` filter for a thread-subtree read. +/// `kinds` is required to prove the filter cannot match p-gated events; without +/// it, relay authorization rejects this otherwise kindless query. fn build_thread_replies_filter( root_event_id: &str, channel_id: Option<&str>, @@ -324,6 +305,7 @@ fn build_thread_replies_filter( // defaults it to a deep-but-bounded value so nested replies aren't dropped. filter.insert("depth_limit".to_string(), serde_json::json!(depth_limit)); filter.insert("limit".to_string(), serde_json::json!(cap)); + filter.insert("include_aux".to_string(), serde_json::json!(true)); if let Some(cid) = channel_id { filter.insert("#h".to_string(), serde_json::json!([cid])); } @@ -435,7 +417,7 @@ pub async fn get_event(event_id: String, state: State<'_, AppState>) -> Result, + root_event_id: Option, media_tags: Option>>, emoji_tags: Option>>, mention_tags: Option>>, @@ -483,6 +466,9 @@ pub async fn send_channel_message( if sent_from_thread_tag.is_some() && kind_num != buzz_core_pkg::kind::KIND_STREAM_MESSAGE { return Err("sent-from-thread provenance requires a stream message".into()); } + if root_event_id.is_some() && parent_event_id.is_none() { + return Err("root_event_id requires parent_event_id".into()); + } let mut resolved_root: Option = None; @@ -498,8 +484,14 @@ pub async fn send_channel_message( let parent_id = parent_event_id .as_deref() .ok_or("forum comment requires parent_event_id")?; - let thread_ref = - resolve_thread_ref(parent_id, &state, &relay_base, Some(&signing_keys)).await?; + let thread_ref = thread_ref( + parent_id, + root_event_id.as_deref(), + &state, + &relay_base, + Some(&signing_keys), + ) + .await?; resolved_root = Some(thread_ref.root_event_id.to_hex()); events::build_forum_comment( channel_uuid, @@ -513,8 +505,14 @@ pub async fn send_channel_message( _ => { let thread_ref = match parent_event_id.as_deref() { Some(pid) => { - let tr = - resolve_thread_ref(pid, &state, &relay_base, Some(&signing_keys)).await?; + let tr = thread_ref( + pid, + root_event_id.as_deref(), + &state, + &relay_base, + Some(&signing_keys), + ) + .await?; resolved_root = Some(tr.root_event_id.to_hex()); Some(tr) } diff --git a/desktop/src-tauri/src/commands/messages/thread_ref.rs b/desktop/src-tauri/src/commands/messages/thread_ref.rs index 97a03fdad5b..8ec82beebb7 100644 --- a/desktop/src-tauri/src/commands/messages/thread_ref.rs +++ b/desktop/src-tauri/src/commands/messages/thread_ref.rs @@ -1,4 +1,4 @@ -use nostr::EventId; +use nostr::{EventId, Keys}; use crate::{ app_state::AppState, @@ -6,6 +6,38 @@ use crate::{ relay::{query_relay_at, query_relay_at_with_keys}, }; +/// Build a thread reference from a renderer-supplied root and parent. +/// +/// Both IDs are parsed before signing. This path intentionally performs no +/// relay query: the renderer supplies a root only when the parent is already +/// present in its cache and the root can be read from that event's NIP-10 tags. +pub(super) fn provided_thread_ref( + root_event_id: &str, + parent_event_id: &str, +) -> Result { + let root_event_id = + EventId::from_hex(root_event_id).map_err(|e| format!("invalid root event ID: {e}"))?; + let parent_event_id = + EventId::from_hex(parent_event_id).map_err(|e| format!("invalid parent event ID: {e}"))?; + Ok(events::ThreadRef { + root_event_id, + parent_event_id, + }) +} + +pub(super) async fn thread_ref( + parent_event_id: &str, + root_event_id: Option<&str>, + state: &AppState, + api_base_url: &str, + signing_keys: Option<&Keys>, +) -> Result { + match root_event_id { + Some(root_event_id) => provided_thread_ref(root_event_id, parent_event_id), + None => resolve_thread_ref(parent_event_id, state, api_base_url, signing_keys).await, + } +} + /// Fetch a parent event and extract the thread root from its NIP-10 e-tags. /// /// Reads through the explicit `api_base_url` the calling command resolved — diff --git a/desktop/src-tauri/src/commands/messages_tests.rs b/desktop/src-tauri/src/commands/messages_tests.rs index c0ad03d936b..dc7c0f4b5a2 100644 --- a/desktop/src-tauri/src/commands/messages_tests.rs +++ b/desktop/src-tauri/src/commands/messages_tests.rs @@ -171,6 +171,7 @@ fn thread_replies_filter_carries_non_p_gated_kinds_to_clear_the_gate() { assert_eq!(filter["#e"], serde_json::json!(["root-hex"])); assert_eq!(filter["depth_limit"], serde_json::json!(64)); assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); + assert_eq!(filter["include_aux"], serde_json::json!(true)); } #[test] @@ -224,3 +225,14 @@ fn legacy_managed_agent_auth_tag_skips_self_attestation() { assert_eq!(tag, None); } + +#[test] +fn provided_thread_ref_validates_and_preserves_root_and_parent() { + let root = "11".repeat(32); + let parent = "22".repeat(32); + let thread_ref = thread_ref::provided_thread_ref(&root, &parent) + .expect("valid 64-hex event ids should be accepted"); + assert_eq!(thread_ref.root_event_id.to_hex(), root); + assert_eq!(thread_ref.parent_event_id.to_hex(), parent); + assert!(thread_ref::provided_thread_ref("not-hex", &parent).is_err()); +} diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index ff022ab522b..63db15cbe46 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -90,6 +90,18 @@ type MessageQueryContext = { const CHANNEL_TIMELINE_KINDS = new Set(CHANNEL_TIMELINE_CONTENT_KINDS); const CHANNEL_AUX_KINDS = new Set(CHANNEL_AUX_EVENT_KINDS); +export function resolveCachedReplyRootId( + parentEventId: string, + messageCaches: readonly RelayEvent[][], +): string | null { + for (const messages of messageCaches) { + if (messages.some((event) => event.id === parentEventId)) { + return resolveReplyRootId(parentEventId, messages); + } + } + return null; +} + export function createOptimisticMessage( channelId: string, content: string, @@ -551,6 +563,17 @@ export function useSendMessageMutation( queryClient.getQueryData( channelMessagesKey(effectiveChannel.id), ) ?? []; + const threadCaches = queryClient + .getQueriesData({ + queryKey: ["thread-replies", effectiveChannel.id], + }) + .flatMap(([, events]) => (events ? [events] : [])); + const suppliedRootEventId = parentEventId + ? resolveCachedReplyRootId(parentEventId, [ + cachedMessages, + ...threadCaches, + ]) + : null; const result = await sendChannelMessage( effectiveChannel.id, content, @@ -562,6 +585,9 @@ export function useSendMessageMutation( mentionTags, linkPreviewTags, sentFromThreadTag, + undefined, + undefined, + suppliedRootEventId, ); // Build tags matching relay-emitted shape: h, author p, mention ps, reply es, imeta, emoji. @@ -572,7 +598,7 @@ export function useSendMessageMutation( effectiveChannel.id, identity.pubkey, parentEventId, - resolveReplyRootId(parentEventId, cachedMessages), + result.rootEventId, recipientPubkeys, ) : []; diff --git a/desktop/src/features/messages/lib/sendChannelBinding.test.mjs b/desktop/src/features/messages/lib/sendChannelBinding.test.mjs index 421799d3734..bfe66afcff1 100644 --- a/desktop/src/features/messages/lib/sendChannelBinding.test.mjs +++ b/desktop/src/features/messages/lib/sendChannelBinding.test.mjs @@ -24,6 +24,7 @@ import test from "node:test"; import { createOptimisticMessage, + resolveCachedReplyRootId, resolveEffectiveChannel, resolveSendChannel, resolveThreadReplyTarget, @@ -322,3 +323,23 @@ test("resolveThreadReplyTarget_nullContext_noLiveRefs_returnsNull", () => { assert.strictEqual(result, null); }); + +test("resolveCachedReplyRootId sends only roots proven by a cached parent", () => { + const rootId = "1".repeat(64); + const parentId = "2".repeat(64); + const parent = { + id: parentId, + pubkey: IDENTITY.pubkey, + kind: 9, + created_at: 1, + content: "parent", + tags: [ + ["e", rootId, "", "root"], + ["e", rootId, "", "reply"], + ], + sig: "", + }; + + assert.equal(resolveCachedReplyRootId(parentId, [[], [parent]]), rootId); + assert.equal(resolveCachedReplyRootId(parentId, [[], []]), null); +}); diff --git a/desktop/src/features/messages/useThreadReplies.test.mjs b/desktop/src/features/messages/useThreadReplies.test.mjs index 48896c5c517..53399cd2f39 100644 --- a/desktop/src/features/messages/useThreadReplies.test.mjs +++ b/desktop/src/features/messages/useThreadReplies.test.mjs @@ -1,30 +1,15 @@ import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; import test from "node:test"; -import { collectThreadAuxMessageIds } from "./useThreadReplies.ts"; - -const ROOT_ID = "1".repeat(64); -const REPLY_ID = "2".repeat(64); - -function reply(id = REPLY_ID) { - return { - id, - pubkey: "a".repeat(64), - kind: 9, - created_at: 1_700_000_000, - content: "reply", - tags: [["e", ROOT_ID]], - sig: "sig", - }; -} - -test("thread aux hydration includes the root when there are no replies", () => { - assert.deepEqual(collectThreadAuxMessageIds(ROOT_ID, []), [ROOT_ID]); -}); - -test("thread aux hydration includes and deduplicates root and reply ids", () => { - assert.deepEqual( - collectThreadAuxMessageIds(ROOT_ID, [reply(), reply(ROOT_ID)]), - [ROOT_ID, REPLY_ID], +test("thread replies trust the relay-provided aux closure", async () => { + const source = await readFile( + new URL("./useThreadReplies.ts", import.meta.url), + "utf8", + ); + assert.doesNotMatch( + source, + /withThreadAux|fetchStructuralAuxForMessages|fetchAuxEventsByReference/, ); + assert.match(source, /replies\.push\(\.\.\.response\.events\)/); }); diff --git a/desktop/src/features/messages/useThreadReplies.ts b/desktop/src/features/messages/useThreadReplies.ts index bb1c2909b68..4b7d7338acd 100644 --- a/desktop/src/features/messages/useThreadReplies.ts +++ b/desktop/src/features/messages/useThreadReplies.ts @@ -5,76 +5,16 @@ import { useQueryClient, } from "@tanstack/react-query"; -import { - collectMessageIdsForAuxBackfill, - fetchStructuralAuxForMessages, -} from "@/features/messages/lib/auxBackfill"; import { threadRepliesKey, sortMessages, } from "@/features/messages/lib/messageQueryKeys"; -import { relayClient } from "@/shared/api/relayClient"; -import { buildChannelReactionAuxFilter } from "@/shared/api/relayChannelFilters"; import { getThreadReplies } from "@/shared/api/tauri"; import type { Channel, RelayEvent, ThreadCursor } from "@/shared/api/types"; const THREAD_PAGE_LIMIT = 200; const MAX_THREAD_PAGES = 500; -/** - * Append the structural aux closure (edits/deletions) for the fetched replies. - * The server thread-subtree query resolves deletions itself but omits - * kind:40003 edits, so a bare refetch would render every edited reply with its - * original text. Best-effort: an aux failure logs and returns the replies - * unadorned rather than failing the whole thread load. - */ -async function fetchThreadAuxBestEffort( - label: string, - channelId: string, - fetchAux: () => Promise, -): Promise { - try { - return await fetchAux(); - } catch (error) { - console.error( - `Failed to backfill thread reply ${label} for channel`, - channelId, - error, - ); - return []; - } -} - -export function collectThreadAuxMessageIds( - threadRootId: string, - replies: RelayEvent[], -): string[] { - return [ - ...new Set([threadRootId, ...collectMessageIdsForAuxBackfill(replies)]), - ]; -} - -async function withThreadAux( - channelId: string, - threadRootId: string, - replies: RelayEvent[], -): Promise { - const messageIds = collectThreadAuxMessageIds(threadRootId, replies); - const [structuralAux, reactions] = await Promise.all([ - fetchThreadAuxBestEffort("structural aux", channelId, () => - fetchStructuralAuxForMessages(channelId, messageIds), - ), - fetchThreadAuxBestEffort("reactions", channelId, () => - relayClient.fetchAuxEventsByReference( - channelId, - messageIds, - buildChannelReactionAuxFilter, - ), - ), - ]); - return sortMessages([...replies, ...structuralAux, ...reactions]); -} - async function loadThreadReplies( queryClient: QueryClient, channelId: string, @@ -92,12 +32,11 @@ async function loadThreadReplies( }); replies.push(...response.events); if (!response.nextCursor) { - const fetched = await withThreadAux(channelId, rootId, replies); const current = queryClient.getQueryData(queryKey) ?? []; const receivedInFlight = current.filter( (event) => !idsAtStart.has(event.id), ); - return sortMessages([...fetched, ...receivedInFlight]); + return sortMessages([...replies, ...receivedInFlight]); } cursor = response.nextCursor; } @@ -123,7 +62,7 @@ export function useThreadReplies( if (!activeChannel || !openThreadRootId) return []; return loadThreadReplies(queryClient, activeChannel.id, openThreadRootId); }, - staleTime: 0, + staleTime: 30_000, gcTime: 60 * 60 * 1_000, }); } @@ -145,7 +84,7 @@ export function useThreadRepliesForRoots( queryKey: threadRepliesKey(channelId, rootId), enabled: activeChannel !== null && activeChannel.channelType !== "forum", queryFn: () => loadThreadReplies(queryClient, channelId, rootId), - staleTime: 0, + staleTime: 30_000, gcTime: 60 * 60 * 1_000, })), combine: (results) => ({ diff --git a/desktop/src/shared/api/tauriMessages.ts b/desktop/src/shared/api/tauriMessages.ts index 4abe03ee09b..965e498b411 100644 --- a/desktop/src/shared/api/tauriMessages.ts +++ b/desktop/src/shared/api/tauriMessages.ts @@ -15,6 +15,7 @@ export async function sendChannelMessage( sentFromThreadTag?: string[], expectedRelayUrl?: string, expectedSignerPubkey?: string, + rootEventId?: string | null, ): Promise { const response = await invokeTauri( "send_channel_message", @@ -22,6 +23,7 @@ export async function sendChannelMessage( channelId, content, parentEventId, + rootEventId: rootEventId ?? null, mediaTags: mediaTags ?? null, emojiTags: emojiTags ?? null, mentionTags: mentionTags ?? null, diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 63f3a4ba385..56e18e23aae 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -9577,6 +9577,7 @@ async function handleSendChannelMessage( channelId: string; content: string; parentEventId?: string | null; + rootEventId?: string | null; kind?: number | null; mentionPubkeys?: string[]; mediaTags?: string[][] | null; @@ -9699,7 +9700,8 @@ async function handleSendChannelMessage( parentEventId: null, rootEventId: null, }; - const rootEventId = parentThread.rootEventId ?? args.parentEventId; + const rootEventId = + args.rootEventId ?? parentThread.rootEventId ?? args.parentEventId; const depth = parentEvent ? (() => { let currentEvent: RelayEvent | undefined = parentEvent; @@ -9758,7 +9760,7 @@ async function handleSendChannelMessage( args.channelId, relayIdentity.pubkey, args.parentEventId, - args.parentEventId, + args.rootEventId ?? args.parentEventId, args.mentionPubkeys, ) : buildTopLevelMessageTags( @@ -9776,8 +9778,12 @@ async function handleSendChannelMessage( return { event_id: result.event_id, parent_event_id: args.parentEventId ?? null, - root_event_id: args.parentEventId ?? null, - depth: args.parentEventId ? 1 : 0, + root_event_id: args.rootEventId ?? args.parentEventId ?? null, + depth: args.parentEventId + ? args.rootEventId && args.rootEventId !== args.parentEventId + ? 2 + : 1 + : 0, created_at: Math.floor(Date.now() / 1000), }; } From 5e8b6c6c3c1e9744e0fe10c08de49c09773f9d8b Mon Sep 17 00:00:00 2001 From: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 12:00:07 -0400 Subject: [PATCH 04/13] fix(desktop): tolerate corrupt channel cache rows Co-authored-by: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz> Signed-off-by: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz> --- desktop/src-tauri/src/channel_head_cache.rs | 38 +++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/desktop/src-tauri/src/channel_head_cache.rs b/desktop/src-tauri/src/channel_head_cache.rs index 2cf2b7ef5ba..f84c534d30c 100644 --- a/desktop/src-tauri/src/channel_head_cache.rs +++ b/desktop/src-tauri/src/channel_head_cache.rs @@ -154,8 +154,13 @@ fn load_from_path( for row in rows { let (channel_id, events_json, saved_at, last_visited_at) = row.map_err(|error| format!("read channel-head cache row: {error}"))?; - let events = serde_json::from_str(&events_json) - .map_err(|error| format!("decode channel-head cache row {channel_id}: {error}"))?; + let events = match serde_json::from_str(&events_json) { + Ok(events) => events, + Err(error) => { + eprintln!("skipping corrupt channel-head cache row {channel_id}: {error}"); + continue; + } + }; entries.push(ChannelHeadEntry { channel_id, events, @@ -371,6 +376,35 @@ mod tests { assert_eq!(count, 0); } + #[test] + fn skips_corrupt_rows_without_blanketing_good_entries() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("channel-head-cache.db"); + store_at( + &path, + &scope(), + "good-channel", + &[serde_json::json!({"id":"good-event"})], + 1_000, + ) + .unwrap(); + let conn = open_db(&path).unwrap(); + conn.execute( + "INSERT INTO channel_head VALUES(?1, 'bad-channel', 'not-json', 1, 1001, 1001)", + [scope().key()], + ) + .unwrap(); + drop(conn); + + let entries = load_from_path(&path, &scope(), 12).unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].channel_id, "good-channel"); + assert_eq!( + entries[0].events, + vec![serde_json::json!({"id":"good-event"})] + ); + } + #[test] fn schema_mismatch_recreates_cache() { let directory = tempfile::tempdir().unwrap(); From d7b45f4ee114443df3075c9d719afe1889fca688 Mon Sep 17 00:00:00 2001 From: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 12:20:32 -0400 Subject: [PATCH 05/13] docs(relay): describe thread auxiliary closure Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> --- docs/bridge-channel-window.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/bridge-channel-window.md b/docs/bridge-channel-window.md index 42f1d82835e..7dfeea0da1c 100644 --- a/docs/bridge-channel-window.md +++ b/docs/bridge-channel-window.md @@ -126,7 +126,9 @@ Both kinds are relay-only: client submission is rejected at ingest. - Reconnect refetches page 0 and re-arms the live subscription (`since: now`); deeper pages need no repair path. - Replies never enter the channel timeline; the thread panel uses the - existing `thread_cursor` surface (#1418). + existing `thread_cursor` surface (#1418). Thread filters may opt into + `include_aux` to append the same authorized two-hop reactions, edits, and + deletions closure as a channel-window response. ## Siblings From 7acbf951bdcd5a59b30b713174b61c0fa737e22d Mon Sep 17 00:00:00 2001 From: Max Date: Sat, 22 Aug 2026 12:51:27 -0400 Subject: [PATCH 06/13] test(desktop): verify persisted head restart paint Co-authored-by: Max Signed-off-by: Max --- desktop/playwright.config.ts | 1 + .../features/channels/ui/ChannelScreen.tsx | 7 +- desktop/src/features/messages/hooks.ts | 2 +- .../messages/lib/channelHeadCache.test.mjs | 125 ++++++++++++++++-- .../features/messages/lib/channelHeadCache.ts | 10 +- desktop/src/testing/e2eBridge.ts | 17 ++- .../tests/e2e/channel-head-restart.spec.ts | 88 ++++++++++++ desktop/tests/helpers/bridge.ts | 2 + 8 files changed, 232 insertions(+), 20 deletions(-) create mode 100644 desktop/tests/e2e/channel-head-restart.spec.ts diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index ff8a0e7703b..9099beff69e 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -99,6 +99,7 @@ export default defineConfig({ "**/scroll-history.spec.ts", "**/channel-dense-second-reach.spec.ts", "**/channel-window-mock-paging.spec.ts", + "**/channel-head-restart.spec.ts", "**/live-broadcast-reply-timeline.spec.ts", "**/markdown-parse-cache.spec.ts", "**/overscroll-boundary.spec.ts", diff --git a/desktop/src/features/channels/ui/ChannelScreen.tsx b/desktop/src/features/channels/ui/ChannelScreen.tsx index 6254afd8c71..86193dc53b7 100644 --- a/desktop/src/features/channels/ui/ChannelScreen.tsx +++ b/desktop/src/features/channels/ui/ChannelScreen.tsx @@ -1,4 +1,5 @@ import * as React from "react"; +import { useQueryClient } from "@tanstack/react-query"; import { useAppShell } from "@/app/AppShellContext"; import { useAppNavigation } from "@/app/navigation/useAppNavigation"; import { useActiveChannelHeader } from "@/features/channels/useActiveChannelHeader"; @@ -45,6 +46,7 @@ import { buildMessageComposerEditTarget } from "@/features/messages/lib/draftMen import { formatTimelineMessages } from "@/features/messages/lib/formatTimelineMessages"; import { DeleteMessageConfirmDialog } from "@/features/messages/ui/DeleteMessageConfirmDialog"; import { getThreadReference } from "@/features/messages/lib/threading"; +import { hasPersistedHydratedChannel } from "@/features/messages/lib/channelHeadCache"; import { resolveTimelineLoadingLatch, selectTimelineLoadingState, @@ -95,6 +97,7 @@ export function ChannelScreen({ targetMessageEvents, targetMessageId, }: ChannelScreenProps) { + const queryClient = useQueryClient(); const { goHome } = useAppNavigation(); const { activeCommunity } = useCommunities(); const { @@ -607,7 +610,9 @@ export function ChannelScreen({ isPlaceholderData: messagesQuery.isPlaceholderData, dataLength: messagesQuery.data?.length ?? null, }, - hasSettledThisChannel, + hasSettledThisChannel || + (activeChannelId !== null && + hasPersistedHydratedChannel(queryClient, activeChannelId)), ); const { settledChannelId, isLoading: isTimelineLoading } = resolveTimelineLoadingLatch( diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 63db15cbe46..5059c7cb482 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -598,7 +598,7 @@ export function useSendMessageMutation( effectiveChannel.id, identity.pubkey, parentEventId, - result.rootEventId, + result.rootEventId ?? parentEventId, recipientPubkeys, ) : []; diff --git a/desktop/src/features/messages/lib/channelHeadCache.test.mjs b/desktop/src/features/messages/lib/channelHeadCache.test.mjs index 31d7076d5ef..c31065daaac 100644 --- a/desktop/src/features/messages/lib/channelHeadCache.test.mjs +++ b/desktop/src/features/messages/lib/channelHeadCache.test.mjs @@ -1,12 +1,56 @@ import assert from "node:assert/strict"; -import test from "node:test"; -import { QueryClient } from "@tanstack/react-query"; +import { after, afterEach, before, test } from "node:test"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { JSDOM } from "jsdom"; +import React from "react"; import { consumeHydratedChannel, hydrateChannelHeads, } from "./channelHeadCache.ts"; import { channelMessagesKey } from "./messageQueryKeys.ts"; -import { reconcileFetchedChannelWindow } from "../hooks.ts"; +import { + reconcileFetchedChannelWindow, + useChannelMessagesQuery, +} from "../hooks.ts"; +const dom = new JSDOM("", { + url: "http://localhost", +}); +const channel = { + id: "channel-a", + name: "general", + channelType: "stream", + visibility: "open", + description: "", + topic: null, + purpose: null, + memberCount: 1, + memberPubkeys: [], + lastMessageAt: null, + archivedAt: null, + participants: [], + participantPubkeys: [], + isMember: true, + ttlSeconds: null, + ttlDeadline: null, +}; +let channelWindowCalls = 0; +let channelWindowEvents = []; + +before(() => { + Object.assign(globalThis, { + document: dom.window.document, + HTMLElement: dom.window.HTMLElement, + IS_REACT_ACT_ENVIRONMENT: true, + window: dom.window, + }); +}); + +afterEach(async () => { + const { cleanup } = await import("@testing-library/react"); + cleanup(); +}); + +after(() => dom.window.close()); const channelId = "channel-a"; const root = { id: "a".repeat(64), @@ -38,14 +82,79 @@ function bounds() { }; } function install(entries) { - globalThis.window = { - localStorage: { getItem: () => null }, - __TAURI_INTERNALS__: { - invoke: async (command) => - command === "channel_head_cache_load" ? entries : null, + channelWindowCalls = 0; + channelWindowEvents = [replacement, bounds()]; + window.localStorage.clear(); + window.__TAURI_INTERNALS__ = { + invoke: async (command) => { + if (command === "channel_head_cache_load") return entries; + if (command === "get_channel_window") { + channelWindowCalls += 1; + return channelWindowEvents; + } + return null; }, }; } + +async function mountChannelQuery(client) { + const { renderHook, waitFor } = await import("@testing-library/react"); + const view = renderHook(() => useChannelMessagesQuery(channel), { + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client }, children), + }); + await waitFor(() => assert.equal(view.result.current.isSuccess, true)); + return view; +} +test("mount fetches cold and prefetched channels but consumes hydrated data", async () => { + install([]); + const coldClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const coldView = await mountChannelQuery(coldClient); + assert.equal(channelWindowCalls, 1); + + install([]); + const prefetchedClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + prefetchedClient.setQueryData(channelMessagesKey(channelId), [root], { + updatedAt: 0, + }); + const prefetchedView = await mountChannelQuery(prefetchedClient); + assert.equal(channelWindowCalls, 1); + + install([ + { channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }, + ]); + const hydratedClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + await hydrateChannelHeads(hydratedClient, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + const hydratedView = await mountChannelQuery(hydratedClient); + assert.equal(channelWindowCalls, 0); + assert.deepEqual(hydratedView.result.current.data, [root]); + + await hydratedClient.invalidateQueries({ + queryKey: channelMessagesKey(channelId), + exact: true, + refetchType: "active", + }); + assert.equal(channelWindowCalls, 1); + assert.deepEqual(hydratedClient.getQueryData(channelMessagesKey(channelId)), [ + replacement, + ]); + hydratedView.unmount(); + coldView.unmount(); + prefetchedView.unmount(); + coldClient.clear(); + prefetchedClient.clear(); + hydratedClient.clear(); +}); + test("hydrates stale data and consumes its mount gate once", async () => { install([ { channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }, diff --git a/desktop/src/features/messages/lib/channelHeadCache.ts b/desktop/src/features/messages/lib/channelHeadCache.ts index 409543be181..90881cbe186 100644 --- a/desktop/src/features/messages/lib/channelHeadCache.ts +++ b/desktop/src/features/messages/lib/channelHeadCache.ts @@ -11,6 +11,7 @@ import { } from "./channelWindowStore"; import { reconcileChannelWindowMessages } from "./channelWindowReconciliation"; const hydratedChannels = new WeakMap>(); +const persistedHydratedChannels = new WeakMap>(); const cacheScopes = new WeakMap(); export function isChannelHeadCacheEnabled(): boolean { if (typeof window === "undefined") return false; @@ -31,11 +32,11 @@ export function consumeHydratedChannel( if (channels.size === 0) hydratedChannels.delete(queryClient); return true; } -export function hasHydratedChannel( +export function hasPersistedHydratedChannel( queryClient: QueryClient, channelId: string, ): boolean { - return hydratedChannels.get(queryClient)?.has(channelId) ?? false; + return persistedHydratedChannels.get(queryClient)?.has(channelId) ?? false; } export async function hydrateChannelHeads( queryClient: QueryClient, @@ -72,5 +73,8 @@ export async function hydrateChannelHeads( ); } } - if (hydrated.size > 0) hydratedChannels.set(queryClient, hydrated); + if (hydrated.size > 0) { + hydratedChannels.set(queryClient, hydrated); + persistedHydratedChannels.set(queryClient, new Set(hydrated)); + } } diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 56e18e23aae..3252a025c0f 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -370,6 +370,8 @@ type E2eConfig = { /** Delay (ms) applied to continuation channel-window requests so e2e * tests can observe the in-flight prepend window. 0/undefined = instant. */ channelWindowDelayMs?: number; + /** Delay (ms) applied to newest-page channel-window requests. */ + channelHeadDelayMs?: number; profileReadDelayMs?: number; profileReadError?: string; /** Override whether get_profile reports a real kind:0 event. */ @@ -5546,19 +5548,20 @@ async function handleGetChannelWindow( return relayQuery(config, [filter]); }; - if (!args.cursor) { - return execute(); - } - const probe = window as unknown as { __CHANNEL_WINDOW_FETCH_COUNT__?: number; __CHANNEL_WINDOW_INFLIGHT__?: number; __CHANNEL_WINDOW_INFLIGHT_PEAK__?: number; }; - probe.__CHANNEL_WINDOW_FETCH_COUNT__ = - (probe.__CHANNEL_WINDOW_FETCH_COUNT__ ?? 0) + 1; + if (args.cursor !== null) { + probe.__CHANNEL_WINDOW_FETCH_COUNT__ = + (probe.__CHANNEL_WINDOW_FETCH_COUNT__ ?? 0) + 1; + } - const delayMs = getConfig()?.mock?.channelWindowDelayMs ?? 0; + const delayMs = + args.cursor === null + ? (getConfig()?.mock?.channelHeadDelayMs ?? 0) + : (getConfig()?.mock?.channelWindowDelayMs ?? 0); if (delayMs <= 0) { return execute(); } diff --git a/desktop/tests/e2e/channel-head-restart.spec.ts b/desktop/tests/e2e/channel-head-restart.spec.ts new file mode 100644 index 00000000000..4bf082786bb --- /dev/null +++ b/desktop/tests/e2e/channel-head-restart.spec.ts @@ -0,0 +1,88 @@ +import { expect, test } from "@playwright/test"; + +import { installMockBridge } from "../helpers/bridge"; + +const PERSISTED_ONLY = "persisted restart head"; + +test("restart paints a persisted head before the single authoritative refresh", async ({ + page, +}) => { + await installMockBridge(page); + await page.goto("/"); + await page.waitForFunction( + () => typeof window.__BUZZ_E2E_EMIT_MOCK_MESSAGE__ === "function", + ); + await page.evaluate((content) => { + window.__BUZZ_E2E_EMIT_MOCK_MESSAGE__?.({ + channelName: "general", + content, + }); + }, PERSISTED_ONLY); + + await page.getByTestId("channel-general").click(); + await expect(page.getByText(PERSISTED_ONLY)).toBeVisible(); + await expect + .poll(() => + page.evaluate( + () => + (window.__BUZZ_E2E_COMMANDS__ ?? []).filter( + (command) => command === "channel_head_cache_store", + ).length, + ), + ) + .toBeGreaterThan(0); + + const persistedCache = await page.evaluate(() => + Object.fromEntries( + Object.entries(window.localStorage).filter(([key]) => + key.startsWith("buzz-e2e-channel-head:"), + ), + ), + ); + expect(Object.keys(persistedCache)).toHaveLength(1); + await page.addInitScript((cache) => { + for (const [key, value] of Object.entries(cache)) { + window.localStorage.setItem(key, value); + } + const testWindow = window as Window & { + __BUZZ_E2E__?: { mock?: Record }; + }; + testWindow.__BUZZ_E2E__ = { + ...testWindow.__BUZZ_E2E__, + mock: { + ...testWindow.__BUZZ_E2E__?.mock, + channelHeadDelayMs: 5_000, + }, + }; + }, persistedCache); + await page.reload(); + await expect(page.getByTestId("channel-general")).toBeVisible(); + const restartedCallsBeforeOpen = await page.evaluate( + () => + (window.__BUZZ_E2E_COMMANDS__ ?? []).filter( + (command) => command === "get_channel_window", + ).length, + ); + await page.getByTestId("channel-general").click(); + + // The mock relay fetch is held for 5s. Seeing this row inside 2s proves + // restart hydration painted persisted data rather than waiting for the relay. + await expect(page.getByText(PERSISTED_ONLY)).toBeVisible({ timeout: 2_000 }); + await expect + .poll(() => + page.evaluate( + (callsBeforeOpen) => + (window.__BUZZ_E2E_COMMANDS__ ?? []).filter( + (command) => command === "get_channel_window", + ).length - callsBeforeOpen, + restartedCallsBeforeOpen, + ), + ) + .toBe(1); + + // Reload resets the mock relay store, so the authoritative refresh omits the + // persisted-only row and must replace it wholesale. + await expect(page.getByText(PERSISTED_ONLY)).toHaveCount(0, { + timeout: 8_000, + }); +}); diff --git a/desktop/tests/helpers/bridge.ts b/desktop/tests/helpers/bridge.ts index ed94e6b1767..2637b94a808 100644 --- a/desktop/tests/helpers/bridge.ts +++ b/desktop/tests/helpers/bridge.ts @@ -304,6 +304,8 @@ type MockBridgeOptions = { usersBatchDelayMs?: number; /** Delay (ms) for older-history fetches; see e2eBridge mock config. */ channelWindowDelayMs?: number; + /** Delay (ms) for newest-page fetches; see e2eBridge mock config. */ + channelHeadDelayMs?: number; profileReadDelayMs?: number; profileReadError?: string; /** Override whether get_profile reports a real kind:0 event. */ From bcfe04e2ff79852cc4d1fa0a87c9a2f02bd0eaaa Mon Sep 17 00:00:00 2001 From: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 14:00:11 -0400 Subject: [PATCH 07/13] fix(desktop): refetch threads when reopened Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz> --- desktop/src/features/messages/useThreadReplies.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/desktop/src/features/messages/useThreadReplies.ts b/desktop/src/features/messages/useThreadReplies.ts index 4b7d7338acd..25a6b68986b 100644 --- a/desktop/src/features/messages/useThreadReplies.ts +++ b/desktop/src/features/messages/useThreadReplies.ts @@ -62,7 +62,7 @@ export function useThreadReplies( if (!activeChannel || !openThreadRootId) return []; return loadThreadReplies(queryClient, activeChannel.id, openThreadRootId); }, - staleTime: 30_000, + staleTime: 0, gcTime: 60 * 60 * 1_000, }); } @@ -84,7 +84,7 @@ export function useThreadRepliesForRoots( queryKey: threadRepliesKey(channelId, rootId), enabled: activeChannel !== null && activeChannel.channelType !== "forum", queryFn: () => loadThreadReplies(queryClient, channelId, rootId), - staleTime: 30_000, + staleTime: 0, gcTime: 60 * 60 * 1_000, })), combine: (results) => ({ From 8133d70bb39c5cca0d915c43c9028003be3af489 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 14:47:22 -0400 Subject: [PATCH 08/13] test(desktop): wait for relay connect before restart-close check The service-restart backoff test waited for the channel list to paint before firing the 1012 close. The channel list is Tauri-backed and renders before the websocket is up, so with three rejected dials the session was still in its backoff loop when the test looked for a socket to close and found none. Now that preconnect dials immediately instead of waiting for requestIdleCallback, the first rejected attempt happens early enough to expose the gap deterministically. Wait for the connected state instead. The test still fails with the 1012 backoff reset disabled, so it guards the same behavior. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- desktop/tests/e2e/relay-reconnect.spec.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/desktop/tests/e2e/relay-reconnect.spec.ts b/desktop/tests/e2e/relay-reconnect.spec.ts index af573373d36..7868e577dab 100644 --- a/desktop/tests/e2e/relay-reconnect.spec.ts +++ b/desktop/tests/e2e/relay-reconnect.spec.ts @@ -368,9 +368,16 @@ test("service restart close resets accumulated backoff", async ({ page }) => { websocketConnectErrors: ["down 1", "down 2", "down 3"], }); await page.goto("/"); - await expect(page.getByTestId("channel-general")).toBeVisible({ - timeout: 15_000, - }); + // Three rejected dials put the session deep in its backoff loop before it + // connects; wait for that connect rather than for the Tauri-backed channel + // list, which paints long before the websocket is up. + await expect + .poll( + () => + page.evaluate(() => window.__BUZZ_E2E_GET_RELAY_CONNECTION_STATE__?.()), + { timeout: 15_000 }, + ) + .toBe("connected"); const startedAt = Date.now(); await restartMockWebsockets(page); From 0c492366d61de62dbc01aeae32d5f83390b243b8 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 15:38:08 -0400 Subject: [PATCH 09/13] docs(desktop): update users-batch freshness comments to 10 minutes The B6 change raised the per-pubkey entry check and query staleTime in useUsersBatchQuery from 60s to 10 minutes; two comments still described the 60s window. Comments only. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- desktop/src/features/profile/hooks.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/desktop/src/features/profile/hooks.ts b/desktop/src/features/profile/hooks.ts index 233b1286125..d46d27df9ee 100644 --- a/desktop/src/features/profile/hooks.ts +++ b/desktop/src/features/profile/hooks.ts @@ -284,8 +284,8 @@ export function useUserProfileQuery(pubkey?: string) { // Per-pubkey resolution cache backing `useUsersBatchQuery`'s delta fetch. // `summary: null` records a relay-confirmed miss so unknown pubkeys aren't -// re-requested every page. Entries older than the hook's 60s staleTime are -// treated as unresolved and refetched. +// re-requested every page. Entries older than the hook's 10-minute staleTime +// are treated as unresolved and refetched. export type UsersBatchEntry = { summary: UserProfileSummary | null; fetchedAt: number; @@ -301,7 +301,8 @@ export const usersBatchEntryKey = (pubkey: string) => [ * run re-fetches these profiles from the relay. Must be called anywhere a * specific profile (or a containing `users-batch` query) is invalidated — * otherwise the re-run resolves from the still-fresh-looking entry and - * renders the stale name/avatar for up to the entry's 60s freshness window. + * renders the stale name/avatar for up to the entry's 10-minute freshness + * window. * Synchronous, so callers can evict before awaiting aggregate invalidations. */ export function evictUsersBatchEntries( From 4f06b77709133acc546734d4f8a7504b66cc5354 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 19:13:48 -0400 Subject: [PATCH 10/13] fix(desktop): mount app while channel heads hydrate; always revalidate Three lifecycle regressions in the persisted channel-head cache, from review at 0c492366d: 1. CommunityQueryProvider rendered no children until channel_head_cache_load settled, so the boot splash, AppReady, and relay preconnect all waited on an optional paint cache (blank window up to the 5s SQLite busy timeout). Hydration now starts in the query client's initializer and the app mounts immediately; only useChannelMessagesQuery awaits channelHeadHydration() before choosing the hydrated or cold path, so a channel opened mid-load still paints persisted rows instead of racing a relay fetch. The seed merges into the existing window store rather than replacing it, since the live subscription can overlay events before the load returns. 2. A hydrated channel skips get_channel_window on mount, leaving the post-subscribe refresh as its only authoritative fetch. A rejected subscribeToChannelLive only logged, so the channel stayed stale for the session. The refresh now runs on both settle branches; the reconnect listener re-syncs when the socket recovers. 3. A bounds-only persisted head (zero rows) was marked hydrated, and the ChannelScreen bypass then settled onto an empty placeholder, flashing the empty-channel intro while the relay revalidated. Rowless heads now take the cold path so the skeleton holds, matching the existing timelineLoadingState contract. Tests: three new cases in channelHeadCache.test.mjs (slow-load race, bounds-only head, subscribe failure); each fails with its fix reverted. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- desktop/src/app/App.tsx | 39 +++---- .../features/channels/ui/ChannelScreen.tsx | 3 + desktop/src/features/messages/hooks.ts | 60 +++++++---- .../messages/lib/channelHeadCache.test.mjs | 102 +++++++++++++++++- .../features/messages/lib/channelHeadCache.ts | 48 +++++++-- 5 files changed, 194 insertions(+), 58 deletions(-) diff --git a/desktop/src/app/App.tsx b/desktop/src/app/App.tsx index 23899cfe145..bdb26c1930d 100644 --- a/desktop/src/app/App.tsx +++ b/desktop/src/app/App.tsx @@ -223,29 +223,24 @@ function CommunityQueryProvider({ pubkey: string | null; relayUrl: string | null; }) { - const [queryClient] = useState(createBuzzQueryClient); - const [isHydrated, setIsHydrated] = useState(!pubkey || !relayUrl); + // Seeding persisted channel heads is part of constructing the client, not a + // gate in front of the app: the splash, AppReady, and relay preconnect mount + // immediately, and only the channel query waits on the cache load (see + // channelHeadHydration). It must start here rather than in an effect — + // React Query fires a child's queryFn when it subscribes, before any parent + // effect runs — and StrictMode's dev-only double initializer just issues one + // redundant read on a discarded client. The provider is keyed on the + // community, so one client maps to one {pubkey, relayUrl} scope. + const [queryClient] = useState(() => { + const client = createBuzzQueryClient(); + if (pubkey && relayUrl) { + void hydrateChannelHeads(client, { pubkey, relayUrl }); + } + return client; + }); useEffect(() => setAvatarProfileSyncQueryClient(queryClient), [queryClient]); - useEffect(() => { - let cancelled = false; - if (!pubkey || !relayUrl) { - setIsHydrated(true); - return; - } - void hydrateChannelHeads(queryClient, { pubkey, relayUrl }) - .catch((error) => { - console.warn("Failed to hydrate persisted channel heads", error); - }) - .finally(() => { - if (!cancelled) setIsHydrated(true); - }); - return () => { - cancelled = true; - }; - }, [pubkey, queryClient, relayUrl]); - useEffect(() => { const e2eWindow = window as Window & { __BUZZ_E2E__?: unknown; @@ -264,9 +259,7 @@ function CommunityQueryProvider({ }, [queryClient]); return ( - - {isHydrated ? children : null} - + {children} ); } diff --git a/desktop/src/features/channels/ui/ChannelScreen.tsx b/desktop/src/features/channels/ui/ChannelScreen.tsx index 86193dc53b7..68df9bc05c6 100644 --- a/desktop/src/features/channels/ui/ChannelScreen.tsx +++ b/desktop/src/features/channels/ui/ChannelScreen.tsx @@ -610,6 +610,9 @@ export function ChannelScreen({ isPlaceholderData: messagesQuery.isPlaceholderData, dataLength: messagesQuery.data?.length ?? null, }, + // A persisted head only counts as hydrated when it has rows to paint + // (channelHeadCache.ts), so this bypass never settles onto an empty + // placeholder while the authoritative refresh is still in flight. hasSettledThisChannel || (activeChannelId !== null && hasPersistedHydratedChannel(queryClient, activeChannelId)), diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 5059c7cb482..a3c1e7f172b 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -26,6 +26,7 @@ import { import { reconcileChannelWindowMessages } from "@/features/messages/lib/channelWindowReconciliation"; import { channelHeadCacheScope, + channelHeadHydration, consumeHydratedChannel, } from "@/features/messages/lib/channelHeadCache"; import { storeChannelHeadCache } from "@/shared/api/tauriChannelHeadCache"; @@ -291,6 +292,10 @@ export function useChannelMessagesQuery(channel: Channel | null) { queryKey, queryFn: async ({ signal }) => { if (!channel) throw new Error("No channel selected."); + // Persisted heads seed asynchronously; wait for that seed so a channel + // opened during boot takes the hydrated path instead of racing it with + // a cold relay fetch. + await channelHeadHydration(queryClient); if (consumeHydratedChannel(queryClient, channel.id)) { return queryClient.getQueryData(queryKey) ?? []; } @@ -418,36 +423,45 @@ export function useChannelSubscription(channel: Channel | null) { }); }); + // The live subscription starts at "now", so it cannot close the gap + // between the last page snapshot and subscription establishment. Always + // refresh once subscription setup settles — on success because freshness + // alone is not proof that no relay events landed in that interval, and on + // failure because a hydrated channel has no other authoritative fetch: + // the relay window endpoint may be healthy even when the live socket is + // not, and the reconnect listener above re-syncs when it recovers. + const refreshAfterSubscribe = (outcome: string) => { + if (isDisposed) return; + void refreshNewestWindow().catch((error) => { + if (!isDisposed) { + console.error( + `Failed to refresh channel window after ${outcome}`, + channelId, + error, + ); + } + }); + }; relayClient .subscribeToChannelLive(channelId, (event) => { if (!isDisposed) { appendMessage(event); } }) - .then((dispose) => { - if (isDisposed) { - void dispose(); - return; - } - - cleanup = dispose; - // The live subscription starts at "now", so it cannot close the gap - // between the last page snapshot and subscription establishment. Always - // refresh after the subscription is active; freshness alone is not a - // proof that no relay events landed in that interval. - void refreshNewestWindow().catch((error) => { - if (!isDisposed) { - console.error( - "Failed to refresh channel window after subscribing", - channelId, - error, - ); + .then( + (dispose) => { + if (isDisposed) { + void dispose(); + return; } - }); - }) - .catch((error) => { - console.error("Failed to subscribe to channel", channelId, error); - }); + cleanup = dispose; + refreshAfterSubscribe("subscribing"); + }, + (error) => { + console.error("Failed to subscribe to channel", channelId, error); + refreshAfterSubscribe("subscription failure"); + }, + ); return () => { isDisposed = true; diff --git a/desktop/src/features/messages/lib/channelHeadCache.test.mjs b/desktop/src/features/messages/lib/channelHeadCache.test.mjs index c31065daaac..458d54cfbae 100644 --- a/desktop/src/features/messages/lib/channelHeadCache.test.mjs +++ b/desktop/src/features/messages/lib/channelHeadCache.test.mjs @@ -1,17 +1,20 @@ import assert from "node:assert/strict"; -import { after, afterEach, before, test } from "node:test"; +import { after, afterEach, before, mock, test } from "node:test"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { JSDOM } from "jsdom"; import React from "react"; import { consumeHydratedChannel, + hasPersistedHydratedChannel, hydrateChannelHeads, } from "./channelHeadCache.ts"; import { channelMessagesKey } from "./messageQueryKeys.ts"; import { reconcileFetchedChannelWindow, useChannelMessagesQuery, + useChannelSubscription, } from "../hooks.ts"; +import { relayClient } from "../../../shared/api/relayClient.ts"; const dom = new JSDOM("", { url: "http://localhost", }); @@ -81,13 +84,18 @@ function bounds() { sig: "", }; } -function install(entries) { +function install(entries, { loadDelayMs = 0 } = {}) { channelWindowCalls = 0; channelWindowEvents = [replacement, bounds()]; window.localStorage.clear(); window.__TAURI_INTERNALS__ = { invoke: async (command) => { - if (command === "channel_head_cache_load") return entries; + if (command === "channel_head_cache_load") { + if (loadDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, loadDelayMs)); + } + return entries; + } if (command === "get_channel_window") { channelWindowCalls += 1; return channelWindowEvents; @@ -206,3 +214,91 @@ test("drops malformed entries independently", async () => { assert.equal(client.getQueryData(channelMessagesKey("bad")), undefined); assert.deepEqual(client.getQueryData(channelMessagesKey(channelId)), [root]); }); + +test("a channel mounted during a slow cache load still takes the hydrated path", async () => { + // Carl/#6572 (1): the app no longer waits for the cache before mounting, so + // the channel query must itself wait for the seed instead of racing it with + // a cold relay fetch. + install( + [{ channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }], + { loadDelayMs: 150 }, + ); + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + void hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + const view = await mountChannelQuery(client); + assert.equal(channelWindowCalls, 0); + assert.deepEqual(view.result.current.data, [root]); + view.unmount(); + client.clear(); +}); +test("a bounds-only persisted head is not hydrated", async () => { + // Carl/#6572 (3): zero rows paint nothing, so the channel must take the + // cold path and hold its skeleton rather than flash the empty-channel intro. + install([{ channelId, events: [bounds()], savedAt: 1, lastVisitedAt: 1 }]); + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + await hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + assert.equal(client.getQueryData(channelMessagesKey(channelId)), undefined); + assert.equal(hasPersistedHydratedChannel(client, channelId), false); + const view = await mountChannelQuery(client); + assert.equal(channelWindowCalls, 1); + assert.deepEqual(view.result.current.data, [replacement]); + view.unmount(); + client.clear(); +}); + +test("a hydrated channel still revalidates when live subscription setup fails", async () => { + // Carl/#6572 (2): the hydrated path skips get_channel_window on mount, so + // the post-subscribe refresh is its only authoritative fetch. A rejected + // subscribe must trigger it too, or the channel stays stale all session. + install([ + { channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }, + ]); + mock.method(relayClient, "subscribeToReconnects", () => () => {}); + mock.method(relayClient, "subscribeToChannelLive", () => + Promise.reject(new Error("socket down")), + ); + const consoleError = mock.method(console, "error", () => {}); + try { + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + await hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + const { renderHook, waitFor } = await import("@testing-library/react"); + const view = renderHook( + () => { + useChannelSubscription(channel); + return useChannelMessagesQuery(channel); + }, + { + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client }, children), + }, + ); + await waitFor(() => assert.equal(channelWindowCalls, 1)); + await waitFor(() => + assert.deepEqual(view.result.current.data, [replacement]), + ); + view.unmount(); + client.clear(); + } finally { + mock.restoreAll(); + } + assert.ok( + consoleError.mock.calls.some( + (call) => call.arguments[0] === "Failed to subscribe to channel", + ), + ); +}); diff --git a/desktop/src/features/messages/lib/channelHeadCache.ts b/desktop/src/features/messages/lib/channelHeadCache.ts index 90881cbe186..0927b40b2c9 100644 --- a/desktop/src/features/messages/lib/channelHeadCache.ts +++ b/desktop/src/features/messages/lib/channelHeadCache.ts @@ -6,10 +6,13 @@ import { import { channelMessagesKey, channelWindowKey } from "./messageQueryKeys"; import { parseChannelWindowResponse } from "./channelWindowResponse"; import { + type ChannelWindowStore, emptyChannelWindowStore, replaceNewestChannelWindow, } from "./channelWindowStore"; import { reconcileChannelWindowMessages } from "./channelWindowReconciliation"; +import type { RelayEvent } from "@/shared/api/types"; +const hydrations = new WeakMap>(); const hydratedChannels = new WeakMap>(); const persistedHydratedChannels = new WeakMap>(); const cacheScopes = new WeakMap(); @@ -23,6 +26,15 @@ export function channelHeadCacheScope( ): ChannelHeadScope | null { return cacheScopes.get(queryClient) ?? null; } +/** + * Resolves once persisted heads have been seeded into this client (or + * immediately when no hydration was started). The channel query awaits this so + * it never races the cache load with a cold relay fetch, while the rest of the + * app mounts without waiting on the optional paint cache. + */ +export function channelHeadHydration(queryClient: QueryClient): Promise { + return hydrations.get(queryClient) ?? Promise.resolve(); +} export function consumeHydratedChannel( queryClient: QueryClient, channelId: string, @@ -38,7 +50,17 @@ export function hasPersistedHydratedChannel( ): boolean { return persistedHydratedChannels.get(queryClient)?.has(channelId) ?? false; } -export async function hydrateChannelHeads( +export function hydrateChannelHeads( + queryClient: QueryClient, + scope: ChannelHeadScope, +): Promise { + const hydration = seedChannelHeads(queryClient, scope).catch((error) => { + console.warn("Failed to hydrate persisted channel heads", error); + }); + hydrations.set(queryClient, hydration); + return hydration; +} +async function seedChannelHeads( queryClient: QueryClient, scope: ChannelHeadScope, ): Promise { @@ -53,17 +75,25 @@ export async function hydrateChannelHeads( entry.channelId, null, ); + // A bounds-only head has nothing to paint; let it take the cold path so + // the skeleton holds until the relay answers instead of flashing the + // empty-channel intro over rows that are still revalidating. + if (page.rows.length === 0) continue; + const windowKey = channelWindowKey(entry.channelId); + const messagesKey = channelMessagesKey(entry.channelId); + // Merge, don't replace: the app mounts while this load is in flight, so + // the live subscription may already have overlaid events on this store. const window = replaceNewestChannelWindow( - emptyChannelWindowStore(), + queryClient.getQueryData(windowKey) ?? + emptyChannelWindowStore(), page, ); - const messages = reconcileChannelWindowMessages(window, []); - queryClient.setQueryData(channelWindowKey(entry.channelId), window, { - updatedAt: 0, - }); - queryClient.setQueryData(channelMessagesKey(entry.channelId), messages, { - updatedAt: 0, - }); + const messages = reconcileChannelWindowMessages( + window, + queryClient.getQueryData(messagesKey) ?? [], + ); + queryClient.setQueryData(windowKey, window, { updatedAt: 0 }); + queryClient.setQueryData(messagesKey, messages, { updatedAt: 0 }); hydrated.add(entry.channelId); } catch (error) { console.warn( From 35834cb313c01edb39f5260a18552e341625b96a Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 19:15:23 -0400 Subject: [PATCH 11/13] fix(relay): drain aux closure hops across the page clamp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both aux-closure call sites (the window path in handle_channel_window_filter and the thread path in query_events_authed) issued each hop as a one-shot `limit: 1000` query. query_events returns newest-first, so a reaction-heavy page past the clamp silently dropped the *oldest* edits and deletions — rendering original or deleted content, not merely losing decoration. build_aux_query no longer sets a limit. A new query_all_pages walks the (created_at, id) keyset the DB already orders by, advancing `until`/`before_id` from the last row of each full page until a short page. Page size is AUX_PAGE_LIMIT (= buzz_db::DEFAULT_MAX_PAGE_LIMIT, one full DB query per page); AUX_MAX_PAGES = 64 bounds a pathological write pattern with a warn + truncated closure instead of a loop. AuxReader is a small enum over the two read paths (the window path pins the request's proved ReadSession; the thread path keeps the routed "bridge_thread_aux" display-read) plus a cfg(test) Fake. An enum rather than an async closure because the closure form tripped a higher-ranked Send bound in the axum handler future. Tests: query_all_pages_drains_past_the_page_clamp emulates the DB's `created_at < until OR (created_at = until AND id > before_id)` cursor with a tied timestamp and asserts every event returned exactly once across 2 full pages + 1 short page (fails with the cursor advance removed); query_all_pages_stops_at_one_short_page; the thread aux query test now asserts limit/until/before_id are unset. docs/bridge-channel-window.md notes each hop is drained server-side. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- crates/buzz-relay/src/api/bridge.rs | 187 ++++++++++++++++++++++++++-- docs/bridge-channel-window.md | 4 +- 2 files changed, 178 insertions(+), 13 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index ee4bd081624..bbfcd8ecfe8 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -393,6 +393,14 @@ const WINDOW_AUX_DELETE_KINDS: [u32; 2] = [ buzz_core::kind::KIND_NIP29_DELETE_EVENT, ]; +/// Page size for one aux-closure hop. Matches the DB clamp +/// (`buzz_db::DEFAULT_MAX_PAGE_LIMIT`) so each page is one full query. +const AUX_PAGE_LIMIT: i64 = buzz_db::DEFAULT_MAX_PAGE_LIMIT; +/// Upper bound on pages drained per hop: 64k aux events referencing one page +/// of rows is far past any real thread; past it we log and stop rather than +/// loop forever against a pathological write pattern. +const AUX_MAX_PAGES: usize = 64; + fn build_aux_query( community: buzz_core::CommunityId, target_ids: Vec, @@ -401,10 +409,67 @@ fn build_aux_query( let mut query = buzz_db::EventQuery::for_community(community); query.kinds = Some(kinds.iter().map(|kind| *kind as i32).collect()); query.e_tags = Some(target_ids); - query.limit = Some(1000); query } +/// Where an aux hop reads from: the window path pins the request's proved +/// read session; the thread path takes the routed display-read fast path. +enum AuxReader<'a> { + Session(&'a mut buzz_db::ReadSession), + Routed(&'a buzz_db::Db, &'static str), + #[cfg(test)] + Fake(&'a mut (dyn FnMut(&buzz_db::EventQuery) -> Vec + Send)), +} + +impl AuxReader<'_> { + async fn fetch( + &mut self, + query: &buzz_db::EventQuery, + ) -> buzz_db::Result> { + match self { + AuxReader::Session(session) => session.query_events(query).await, + AuxReader::Routed(db, path) => db.query_events_routed(path, query).await, + #[cfg(test)] + AuxReader::Fake(fetch) => Ok(fetch(query)), + } + } +} + +/// Drain every event matching `query`, walking the `(created_at, id)` keyset +/// cursor `query_events` already orders by until a short page. An aux hop +/// over a reaction-heavy page can exceed a single page clamp, and because +/// results are newest-first a one-shot query silently drops the *oldest* +/// edits and deletions — rendering original or deleted content, not merely +/// losing decoration. +async fn query_all_pages( + mut query: buzz_db::EventQuery, + page_limit: i64, + reader: &mut AuxReader<'_>, +) -> buzz_db::Result> { + query.limit = Some(page_limit); + let mut events = Vec::new(); + for _ in 0..AUX_MAX_PAGES { + let page = reader.fetch(&query).await?; + let next = if page.len() as i64 >= page_limit { + page.last().map(|se| (se.event.created_at, se.event.id)) + } else { + None + }; + events.extend(page); + let Some((created_at, id)) = next else { + return Ok(events); + }; + query.until = chrono::DateTime::from_timestamp(created_at.as_secs() as i64, 0); + query.before_id = Some(id.to_bytes().to_vec()); + } + tracing::warn!( + pages = AUX_MAX_PAGES, + events = events.len(), + "aux closure hop exceeded page cap; returning truncated closure" + ); + Ok(events) +} + /// Serve one `top_level: true` channel-window filter on the bridge `/query` /// path (docs/bridge-channel-window.md). Appends, in order: row events, the /// aux closure (`include_aux`), `39005` thread-summary overlays @@ -510,10 +575,13 @@ async fn handle_channel_window_filter( for hop_kinds in [&WINDOW_AUX_KINDS[..], &WINDOW_AUX_DELETE_KINDS[..]] { let aux_query = build_aux_query(tenant.community(), std::mem::take(&mut hop_ids), hop_kinds); - let aux_events = session - .query_events(&aux_query) - .await - .map_err(|e| internal_error(&format!("window aux error: {e}")))?; + let aux_events = query_all_pages( + aux_query, + AUX_PAGE_LIMIT, + &mut AuxReader::Session(&mut session), + ) + .await + .map_err(|e| internal_error(&format!("window aux error: {e}")))?; for se in aux_events { if !seen_aux.insert(se.event.id) { continue; @@ -1238,11 +1306,13 @@ async fn query_events_authed( for hop_kinds in [&WINDOW_AUX_KINDS[..], &WINDOW_AUX_DELETE_KINDS[..]] { let aux_query = build_aux_query(tenant.community(), std::mem::take(&mut hop_ids), hop_kinds); - let aux_events = state - .db - .query_events_routed("bridge_thread_aux", &aux_query) - .await - .map_err(|e| internal_error(&format!("thread aux query error: {e}")))?; + let aux_events = query_all_pages( + aux_query, + AUX_PAGE_LIMIT, + &mut AuxReader::Routed(&state.db, "bridge_thread_aux"), + ) + .await + .map_err(|e| internal_error(&format!("thread aux query error: {e}")))?; for se in aux_events { if !seen_aux.insert(se.event.id) || !event_in_accessible_channel(&se, &accessible_channels) @@ -2419,7 +2489,7 @@ mod tests { } #[test] - fn thread_aux_query_targets_root_and_replies_with_full_first_hop() { + fn thread_aux_query_targets_root_and_replies() { let tenant = fresh_tenant("relay.example"); let targets = vec!["root".to_string(), "reply".to_string()]; let query = build_aux_query(tenant.community(), targets.clone(), &WINDOW_AUX_KINDS); @@ -2429,7 +2499,100 @@ mod tests { query.kinds, Some(WINDOW_AUX_KINDS.iter().map(|kind| *kind as i32).collect()) ); - assert_eq!(query.limit, Some(1000)); + assert_eq!(query.limit, None); + assert_eq!(query.until, None); + assert_eq!(query.before_id, None); + } + + fn aux_event(keys: &Keys, created_at: u64, content: &str) -> buzz_core::StoredEvent { + let ev = EventBuilder::new(Kind::Custom(7), content) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(keys) + .unwrap(); + buzz_core::StoredEvent::new(ev, None) + } + + /// Carl/#6572: a one-shot `limit=1000` aux query is newest-first, so the + /// oldest reactions/edits/deletions past the clamp vanished. The paged + /// drain must walk the keyset cursor until a short page and return every + /// event exactly once. + #[tokio::test] + async fn query_all_pages_drains_past_the_page_clamp() { + let keys = Keys::generate(); + // Newest-first store: 5 events, two sharing a second so the id + // tiebreak is exercised. + let mut store = [ + aux_event(&keys, 50, "e"), + aux_event(&keys, 40, "d1"), + aux_event(&keys, 40, "d2"), + aux_event(&keys, 30, "c"), + aux_event(&keys, 10, "a"), + ]; + store.sort_by(|l, r| { + r.event + .created_at + .cmp(&l.event.created_at) + .then(l.event.id.cmp(&r.event.id)) + }); + let expected: Vec<_> = store.iter().map(|se| se.event.id).collect(); + let mut calls = Vec::new(); + + let tenant = fresh_tenant("relay.example"); + let query = build_aux_query(tenant.community(), vec!["root".into()], &WINDOW_AUX_KINDS); + let mut fetch = |q: &buzz_db::EventQuery| { + calls.push((q.limit, q.until, q.before_id.clone())); + // Emulate `query_events_on`: `created_at < until OR + // (created_at = until AND id > before_id)`, newest-first, limit. + let page: Vec<_> = store + .iter() + .filter(|se| match (q.until, q.before_id.as_deref()) { + (Some(until), Some(before)) => { + let ts = se.event.created_at.as_secs() as i64; + ts < until.timestamp() + || (ts == until.timestamp() + && se.event.id.as_bytes().as_slice() > before) + } + _ => true, + }) + .take(q.limit.unwrap() as usize) + .cloned() + .collect(); + page + }; + let events = query_all_pages(query, 2, &mut AuxReader::Fake(&mut fetch)) + .await + .unwrap(); + + assert_eq!( + events.iter().map(|se| se.event.id).collect::>(), + expected + ); + assert_eq!(calls.len(), 3, "2 full pages + 1 short page"); + assert!(calls.iter().all(|(limit, _, _)| *limit == Some(2))); + assert_eq!(calls[0].1, None); + // Second page resumes from the last row of the first (ts 40, larger id). + assert_eq!(calls[1].1.unwrap().timestamp(), 40); + assert_eq!( + calls[1].2.as_deref(), + Some(store[1].event.id.as_bytes().as_slice()) + ); + assert_eq!(calls[2].1.unwrap().timestamp(), 30); + } + + #[tokio::test] + async fn query_all_pages_stops_at_one_short_page() { + let tenant = fresh_tenant("relay.example"); + let query = build_aux_query(tenant.community(), vec!["root".into()], &WINDOW_AUX_KINDS); + let mut calls = 0; + let mut fetch = |_q: &buzz_db::EventQuery| { + calls += 1; + Vec::new() + }; + let events = query_all_pages(query, 1000, &mut AuxReader::Fake(&mut fetch)) + .await + .unwrap(); + assert!(events.is_empty()); + assert_eq!(calls, 1); } #[test] diff --git a/docs/bridge-channel-window.md b/docs/bridge-channel-window.md index 7dfeea0da1c..aec9d19542a 100644 --- a/docs/bridge-channel-window.md +++ b/docs/bridge-channel-window.md @@ -87,7 +87,9 @@ Clients **partition by kind before any cursor math**: 2. **Aux closure** (`include_aux`) — reactions (7), deletions (5, 9005), and edits (40003) targeting the retained rows by `#e`, **plus** deletions targeting those aux events (the transitive second hop, e.g. - a delete-of-a-reaction). One round trip; no client `#e` fan-out. + a delete-of-a-reaction). One round trip; no client `#e` fan-out. Each + hop is drained server-side across the DB page clamp, so the closure is + complete rather than newest-1000. 3. **Thread summaries** (`include_summaries`) — one relay-signed `kind:39005` per row that has replies. 4. **Window bounds** — exactly one relay-signed `kind:39006` per window From 5a5566c0f954ece5130d65ce53ebec11b61eaf83 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sun, 23 Aug 2026 12:02:35 -0400 Subject: [PATCH 12/13] fix(desktop): sequence post-subscribe refresh behind channel head hydration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Carl/#6572 re-review: useChannelMessagesQuery parks on channelHeadHydration() while useChannelSubscription starts concurrently. When subscribeToChannelLive settles before the SQLite head load, refreshAfterSubscribe() invalidates a query that is fetching with no data yet. TanStack dedupes that onto the in-flight fetch (query.fetch only cancels when state.data exists), the seed lands, consumeHydratedChannel() returns the snapshot, and get_channel_window is never called — the channel stays stale until reconnect or remount. refreshChannelWindowMessages now awaits channelHeadHydration() first, and when the query holds seeded data (dataUpdatedAt 0 — only the hydration seed writes that) awaits the snapshot fetch's promise so the mount gate is consumed before invalidating. The refetch is then a distinct authoritative window fetch. Cold and warm channels carry no such marker and dedupe or cancel exactly as before; one call site covers the subscribe-success, subscribe-failure and reconnect paths. Tests (channelHeadCache.test.mjs): - subscribe resolves before a 150 ms cache load → 1 get_channel_window call and the relay row replaces the persisted one. Fails at the previous head with 0 calls. - cold channel with an immediate subscription → still exactly 1 call. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- .../messages/lib/channelHeadCache.test.mjs | 78 +++++++++++++++++++ .../messages/lib/projectChannelWindow.ts | 16 +++- 2 files changed, 93 insertions(+), 1 deletion(-) diff --git a/desktop/src/features/messages/lib/channelHeadCache.test.mjs b/desktop/src/features/messages/lib/channelHeadCache.test.mjs index 458d54cfbae..9fc62331f8f 100644 --- a/desktop/src/features/messages/lib/channelHeadCache.test.mjs +++ b/desktop/src/features/messages/lib/channelHeadCache.test.mjs @@ -302,3 +302,81 @@ test("a hydrated channel still revalidates when live subscription setup fails", ), ); }); + +test("a live subscription that settles before a slow cache load still revalidates", async () => { + // Carl/#6572 re-review: the query is parked on the hydration gate with no + // data yet, so an invalidation issued now dedupes onto that in-flight fetch + // (cancelRefetch only cancels when data exists). The seed then lands and the + // queryFn returns the persisted snapshot — zero authoritative fetches. + install( + [{ channelId, events: [root, bounds()], savedAt: 1, lastVisitedAt: 1 }], + { loadDelayMs: 150 }, + ); + mock.method(relayClient, "subscribeToReconnects", () => () => {}); + mock.method(relayClient, "subscribeToChannelLive", () => + Promise.resolve(async () => {}), + ); + try { + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + void hydrateChannelHeads(client, { + pubkey: "f".repeat(64), + relayUrl: "wss://relay", + }); + const { renderHook, waitFor } = await import("@testing-library/react"); + const view = renderHook( + () => { + useChannelSubscription(channel); + return useChannelMessagesQuery(channel); + }, + { + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client }, children), + }, + ); + await waitFor(() => assert.equal(channelWindowCalls, 1)); + await waitFor(() => + assert.deepEqual(view.result.current.data, [replacement]), + ); + view.unmount(); + client.clear(); + } finally { + mock.restoreAll(); + } +}); + +test("a cold channel with an immediate live subscription fetches the window once", async () => { + // The post-subscribe refresh must still dedupe onto a cold relay fetch that + // is already in flight; only the hydration-parked fetch needs sequencing. + install([]); + mock.method(relayClient, "subscribeToReconnects", () => () => {}); + mock.method(relayClient, "subscribeToChannelLive", () => + Promise.resolve(async () => {}), + ); + try { + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const { renderHook, waitFor } = await import("@testing-library/react"); + const view = renderHook( + () => { + useChannelSubscription(channel); + return useChannelMessagesQuery(channel); + }, + { + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client }, children), + }, + ); + await waitFor(() => + assert.deepEqual(view.result.current.data, [replacement]), + ); + await waitFor(() => assert.equal(view.result.current.isFetching, false)); + assert.equal(channelWindowCalls, 1); + view.unmount(); + client.clear(); + } finally { + mock.restoreAll(); + } +}); diff --git a/desktop/src/features/messages/lib/projectChannelWindow.ts b/desktop/src/features/messages/lib/projectChannelWindow.ts index 81ef3de42d0..9f64cdfbe34 100644 --- a/desktop/src/features/messages/lib/projectChannelWindow.ts +++ b/desktop/src/features/messages/lib/projectChannelWindow.ts @@ -7,6 +7,7 @@ import { type ChannelWindowStore, } from "./channelWindowStore"; import { reconcileChannelWindowMessages } from "./channelWindowReconciliation"; +import { channelHeadHydration } from "./channelHeadCache"; /** Keep the rendered timeline cache aligned with its authoritative window. */ export function projectChannelWindowMessages( @@ -26,8 +27,21 @@ export async function refreshChannelWindowMessages( queryClient: QueryClient, channelId: string, ) { + const queryKey = channelMessagesKey(channelId); + // Sequence behind persisted-head hydration. While the channel query is parked + // on that gate it has no data, so TanStack would dedupe this invalidation + // onto it — and that fetch returns the seeded snapshot, never asking the + // relay. A seeded query is recognisable by data at `dataUpdatedAt` 0; let its + // snapshot fetch settle (consuming the mount gate) before invalidating, so + // the refetch is a distinct authoritative window fetch. Cold and warm + // channels carry no such marker and dedupe/cancel exactly as before. + await channelHeadHydration(queryClient); + const query = queryClient.getQueryCache().find({ queryKey, exact: true }); + if (query?.state.data !== undefined && query.state.dataUpdatedAt === 0) { + await query.promise?.catch(() => {}); + } await queryClient.invalidateQueries({ - queryKey: channelMessagesKey(channelId), + queryKey, exact: true, refetchType: "active", }); From b129231c88f92d95b3fd3ce0fe20204fb2769c5b Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sun, 23 Aug 2026 12:19:25 -0400 Subject: [PATCH 13/13] fix(desktop): let concurrent post-hydration refreshes share one window fetch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Max and Wren (#6572 review of 5a5566c0f): when subscribe settlement and a reconnect both call refreshChannelWindowMessages while the channel query is parked on its hydration-seeded snapshot, both await the same query.promise and both invalidate on wake. The first invalidation starts the authoritative fetch; the second, with TanStack's default cancelRefetch: true, cancels and replaces it — 3 queryFn calls (snapshot + 2 authoritative) where 2 suffice, and the cancelled Tauri invoke still reaches the relay. On the seeded branch only, invalidate with cancelRefetch: false so a second waker joins the in-flight authoritative fetch. Cold and warm channels keep the default: test_canceled_stale_fetch_cannot_overwrite_catch_up_window relies on a catch-up refresh replacing a stale active fetch. Test (projectChannelWindow.test.mjs): seed at updatedAt 0, park the snapshot fetch, call the helper twice, release — exactly 2 requests, the second not aborted, projection shows the relay gap row. Fails at 5a5566c0f with 3. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- .../lib/projectChannelWindow.test.mjs | 53 +++++++++++++++++++ .../messages/lib/projectChannelWindow.ts | 20 ++++--- 2 files changed, 65 insertions(+), 8 deletions(-) diff --git a/desktop/src/features/messages/lib/projectChannelWindow.test.mjs b/desktop/src/features/messages/lib/projectChannelWindow.test.mjs index 14ec110addf..618f3fc9912 100644 --- a/desktop/src/features/messages/lib/projectChannelWindow.test.mjs +++ b/desktop/src/features/messages/lib/projectChannelWindow.test.mjs @@ -363,3 +363,56 @@ test("test_pageless_live_projection_preserves_cached_timeline", () => { assert.deepEqual(contents(harness), ["initial", "live"]); assert.equal(harness.client.getQueryData(harness.messagesKey)[0], cached[0]); }); + +test("test_concurrent_refreshes_after_seeded_snapshot_share_one_authoritative_fetch", async () => { + // Subscribe settlement and a reconnect can both call the helper while the + // channel query is still parked on its hydration-seeded snapshot. Both wake + // on the same promise; the second invalidation must join the first + // authoritative fetch, not cancel and replace it (Max/Wren, #6572 review). + const harness = createHarness(); + const seeded = event("seeded", 100); + harness.client.setQueryData(harness.messagesKey, [seeded], { updatedAt: 0 }); + const requests = []; + const observer = new QueryObserver(harness.client, { + queryKey: harness.messagesKey, + queryFn: async ({ signal }) => { + const previousMessages = harness.client.getQueryData(harness.messagesKey); + let resolveFetch; + const fetch = new Promise((resolve) => { + resolveFetch = resolve; + }); + requests.push({ resolveFetch, signal }); + const events = await fetch; + return reconcileFetchedChannelWindow( + harness.client, + harness.channelId, + events, + previousMessages, + signal, + ); + }, + }); + const unsubscribe = observer.subscribe(() => {}); + try { + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(requests.length, 1); + const first = refreshChannelWindowMessages( + harness.client, + harness.channelId, + ); + const second = refreshChannelWindowMessages( + harness.client, + harness.channelId, + ); + requests[0].resolveFetch(wirePage([seeded])); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(requests.length, 2); + requests[1].resolveFetch(wirePage([event("gap", 110), seeded])); + await Promise.all([first, second]); + assert.equal(requests.length, 2); + assert.equal(requests[1].signal.aborted, false); + assert.deepEqual(contents(harness), ["seeded", "gap"]); + } finally { + unsubscribe(); + } +}); diff --git a/desktop/src/features/messages/lib/projectChannelWindow.ts b/desktop/src/features/messages/lib/projectChannelWindow.ts index 9f64cdfbe34..b16187ce18c 100644 --- a/desktop/src/features/messages/lib/projectChannelWindow.ts +++ b/desktop/src/features/messages/lib/projectChannelWindow.ts @@ -33,17 +33,21 @@ export async function refreshChannelWindowMessages( // onto it — and that fetch returns the seeded snapshot, never asking the // relay. A seeded query is recognisable by data at `dataUpdatedAt` 0; let its // snapshot fetch settle (consuming the mount gate) before invalidating, so - // the refetch is a distinct authoritative window fetch. Cold and warm - // channels carry no such marker and dedupe/cancel exactly as before. + // the refetch is a distinct authoritative window fetch. Concurrent callers + // (subscribe settlement + reconnect) wake on the same promise, so the seeded + // branch must join an authoritative fetch already in flight rather than + // cancel and replace it. Cold and warm channels carry no such marker and + // dedupe/cancel exactly as before. await channelHeadHydration(queryClient); const query = queryClient.getQueryCache().find({ queryKey, exact: true }); - if (query?.state.data !== undefined && query.state.dataUpdatedAt === 0) { + const seeded = + query?.state.data !== undefined && query.state.dataUpdatedAt === 0; + if (seeded) { await query.promise?.catch(() => {}); } - await queryClient.invalidateQueries({ - queryKey, - exact: true, - refetchType: "active", - }); + await queryClient.invalidateQueries( + { queryKey, exact: true, refetchType: "active" }, + { cancelRefetch: !seeded }, + ); projectChannelWindowMessages(queryClient, channelId); }