jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(threads): collapse reply reads and sends

Browse Source 
Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
This commit is contained in:
Wren
2026-08-22 11:40:53 -04:00
parent f1e2e4dc9d
commit 9d7623b3b5
10 changed files with 214 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
crates/buzz-relay/src/api/bridge.rs
View File

@@ -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<String>,
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!(

68
desktop/src-tauri/src/commands/messages.rs
View File

@@ -236,22 +236,11 @@ fn search_messages_limit(limit: Option<u32>) -> 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<S
// ── Writes ──────────────────────────────────────────────────────────────────
mod thread_ref;
use thread_ref::resolve_thread_ref;
use thread_ref::{resolve_thread_ref, thread_ref};
#[tauri::command]
#[allow(clippy::too_many_arguments)]
@@ -443,6 +425,7 @@ pub async fn send_channel_message(
channel_id: String,
content: String,
parent_event_id: Option<String>,
root_event_id: Option<String>,
media_tags: Option<Vec<Vec<String>>>,
emoji_tags: Option<Vec<Vec<String>>>,
mention_tags: Option<Vec<Vec<String>>>,
@@ -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<String> = 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)
}

34
desktop/src-tauri/src/commands/messages/thread_ref.rs
View File

@@ -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<events::ThreadRef, String> {
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<events::ThreadRef, String> {
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 —

12
desktop/src-tauri/src/commands/messages_tests.rs
View File

@@ -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());
}

28
desktop/src/features/messages/hooks.ts
View File

@@ -90,6 +90,18 @@ type MessageQueryContext = {
const CHANNEL_TIMELINE_KINDS = new Set<number>(CHANNEL_TIMELINE_CONTENT_KINDS);
const CHANNEL_AUX_KINDS = new Set<number>(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<RelayEvent[]>(
channelMessagesKey(effectiveChannel.id),
) ?? [];
const threadCaches = queryClient
.getQueriesData<RelayEvent[]>({
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,
)
: [];

21
desktop/src/features/messages/lib/sendChannelBinding.test.mjs
View File

@@ -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);
});

35
desktop/src/features/messages/useThreadReplies.test.mjs
View File

@@ -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\)/);
});

67
desktop/src/features/messages/useThreadReplies.ts
View File

@@ -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<RelayEvent[]>,
): Promise<RelayEvent[]> {
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<RelayEvent[]> {
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<RelayEvent[]>(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) => ({

2
desktop/src/shared/api/tauriMessages.ts
View File

@@ -15,6 +15,7 @@ export async function sendChannelMessage(
sentFromThreadTag?: string[],
expectedRelayUrl?: string,
expectedSignerPubkey?: string,
rootEventId?: string | null,
): Promise<SendChannelMessageResult> {
const response = await invokeTauri<RawSendChannelMessageResult>(
"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,

14
desktop/src/testing/e2eBridge.ts
View File

@@ -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),
};
}