From 4d74e8e0fb71ec95fc122ce560cd7cf70615c83a Mon Sep 17 00:00:00 2001 From: Wintermute <3f1797424fd9ad6653a83665c660517777cd7f8c228c0d5907f49e01537f3ca5@buzz.block.builderlab.xyz> Date: Thu, 13 Aug 2026 23:58:35 -0400 Subject: [PATCH] fix(feed): window the mentions feed per conversation to stop inbox starvation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem The Inbox collapses to a handful of rows the busier your agents get. Reported by Morgan (3 rows spanning 12 hours) and reproduced by Thomas (~10 rows spanning a day and a half). ## Cause The mentions feed was a flat `ORDER BY created_at DESC LIMIT n` over every event p-tagging the user. Clients group those events into conversation rows *after* the cut, so the window is event-shaped while the inbox is conversation-shaped: one chatty DM or thread (e.g. an agent posting many consecutive callback mentions) consumes nearly every slot, then grouping collapses them into a single row. Every conversation older than the window is gone before the UI ever sees it. ## Fix Window the mentions query per conversation in the relay (`build_mentions_query`, crates/buzz-db/src/feed.rs): - candidates: indexed `event_mentions` walk, newest-first, bounded by `FEED_WINDOW_SCAN_CAP` (2000) - keyed: conversation key = `dm:` for DM channels, else the NIP-10 thread root from `thread_metadata` (event's own id for top-level) — mirroring the client's `getInboxConversationId` - ranked: `ROW_NUMBER()` per conversation keeps the newest `FEED_CONVERSATION_EVENT_CAP` (3) events; conversations surface newest-activity-first, so the final `LIMIT` truncates at a conversation boundary A 50-event window now spans at least ~17 distinct conversations instead of potentially 1. No schema change. The desktop `get_feed` mention filter now carries `feed_types: ["mentions"]`, routing it through the relay's feed path (which applies the windowing) instead of the flat generic query. The kinds list is retained as the fallback bound for older relays that ignore `feed_types`. The e2e bridge mirrors the same wire shape. ## Verification - SQL-shape unit test (`mentions_query_windows_per_conversation`) and a Postgres regression (`query_mentions_survives_chatty_conversation_ starvation`): a 40-reply chatty thread plus 6 standalone mentions in a 20-event window — all 7 conversations survive, chatty capped at 3. - Before/after Playwright spec against an isolated relay seeding a 60-message DM burst over 6 standalone mentions (inbox-windowing-screenshots.spec.ts): the flat window shows 1 surviving row; the windowed path shows all 7 conversations. - buzz-db suite, buzz-relay lib suite (879), desktop tauri tests (2414), desktop check/typecheck/test (4775) all green. Co-authored-by: Thomas Petersen Signed-off-by: Thomas Petersen --- crates/buzz-db/src/feed.rs | 253 +++++++++++++++++- desktop/playwright.config.ts | 1 + desktop/src-tauri/src/commands/messages.rs | 8 + desktop/src/testing/e2eBridge.ts | 7 + .../e2e/inbox-windowing-screenshots.spec.ts | 207 ++++++++++++++ 5 files changed, 473 insertions(+), 3 deletions(-) create mode 100644 desktop/tests/e2e/inbox-windowing-screenshots.spec.ts diff --git a/crates/buzz-db/src/feed.rs b/crates/buzz-db/src/feed.rs index 6900e2061c5..22476d6bc2a 100644 --- a/crates/buzz-db/src/feed.rs +++ b/crates/buzz-db/src/feed.rs @@ -28,6 +28,28 @@ /// before the query is issued so the SQL `LIMIT` clause always reflects this cap. pub const FEED_MAX_LIMIT: i64 = 100; +/// Per-conversation cap inside the mentions window. +/// +/// The mentions feed used to be a flat `ORDER BY created_at DESC LIMIT n` +/// over every event p-tagging the user. Clients group those events into +/// conversation rows *after* the cut, so one chatty thread or DM could occupy +/// nearly the whole window and starve every other conversation out of the +/// inbox (observed in production: a single DM held 39 of a 50-event +/// window, collapsing the inbox to 3 rows). Capping each conversation +/// at this many events guarantees a `limit`-row window spans at least +/// `limit / FEED_CONVERSATION_EVENT_CAP` distinct conversations. The client +/// only needs a representative event plus an unread signal per row — opening +/// a row fetches the full thread separately. +pub const FEED_CONVERSATION_EVENT_CAP: i64 = 3; + +/// Upper bound on candidate rows scanned before conversation windowing. +/// +/// Bounds the work of the window function: the candidate CTE walks the +/// indexed `event_mentions` ordering newest-first and stops here. Mentions +/// older than the newest `FEED_WINDOW_SCAN_CAP` mention-events are outside +/// the feed window (they remain reachable via thread/channel queries). +const FEED_WINDOW_SCAN_CAP: i64 = 2000; + use chrono::{DateTime, Utc}; use sqlx::postgres::PgRow; use sqlx::{PgPool, QueryBuilder}; @@ -82,6 +104,22 @@ fn collect_stored_events(rows: Vec) -> Result> { Ok(out) } +/// Build the windowed mentions query. +/// +/// The window is **per-conversation**, not flat: a naive +/// `ORDER BY created_at DESC LIMIT n` lets one chatty thread or DM occupy +/// nearly every slot, starving all other conversations out of the inbox +/// before the client's conversation grouping ever sees them. +/// +/// Shape: a candidate CTE walks the indexed `event_mentions` ordering +/// newest-first (bounded by [`FEED_WINDOW_SCAN_CAP`]), each candidate is +/// keyed by its conversation — `dm:` for DM channels, else the +/// thread root from `thread_metadata` (falling back to the event's own id +/// for top-level events) — mirroring the client's grouping key +/// (`getInboxConversationId`). A `ROW_NUMBER()` window keeps the newest +/// [`FEED_CONVERSATION_EVENT_CAP`] events per conversation, and whole +/// conversations are emitted newest-activity-first so the final `LIMIT` +/// truncates at a conversation boundary instead of mid-window. fn build_mentions_query( community: CommunityId, pubkey_bytes: &[u8], @@ -93,7 +131,8 @@ fn build_mentions_query( let pubkey_hex = hex::encode(pubkey_bytes); let mut qb: QueryBuilder = QueryBuilder::new(format!( - "SELECT {EVENT_COLS} FROM events e \ + "WITH candidates AS ( \ + SELECT {EVENT_COLS} FROM events e \ INNER JOIN event_mentions m ON e.community_id = m.community_id AND e.id = m.event_id \ WHERE e.community_id = " )); @@ -112,8 +151,31 @@ fn build_mentions_query( if let Some(s) = since { qb.push(" AND m.event_created_at >= ").push_bind(s); } - qb.push(" ORDER BY m.event_created_at DESC LIMIT ") - .push_bind(limit); + qb.push(format!( + " ORDER BY m.event_created_at DESC LIMIT {FEED_WINDOW_SCAN_CAP} \ + ), keyed AS ( \ + SELECT c.*, \ + CASE WHEN ch.channel_type = 'dm' THEN 'dm:' || c.channel_id::text \ + ELSE encode(COALESCE(tm.root_event_id, c.id), 'hex') END AS conv_key \ + FROM candidates c \ + LEFT JOIN channels ch ON ch.community_id = " + )); + qb.push_bind(*community.as_uuid()); + qb.push(" AND ch.id = c.channel_id LEFT JOIN thread_metadata tm ON tm.community_id = "); + qb.push_bind(*community.as_uuid()); + qb.push(format!( + " AND tm.event_created_at = c.created_at AND tm.event_id = c.id \ + ), ranked AS ( \ + SELECT k.*, \ + ROW_NUMBER() OVER (PARTITION BY k.conv_key ORDER BY k.created_at DESC, k.id DESC) AS conv_rank, \ + MAX(k.created_at) OVER (PARTITION BY k.conv_key) AS conv_latest \ + FROM keyed k \ + ) \ + SELECT {EVENT_COLS_UNALIASED} FROM ranked \ + WHERE conv_rank <= {FEED_CONVERSATION_EVENT_CAP} \ + ORDER BY conv_latest DESC, conv_key, created_at DESC LIMIT " + )); + qb.push_bind(limit); qb } @@ -374,6 +436,63 @@ mod tests { event } + /// Like [`store_feed_event`], but with a fixed `created_at` and — when the + /// tags carry a NIP-10 root marker — a `thread_metadata` row, mirroring the + /// relay ingest path so the windowed mentions query can resolve the + /// conversation key. + async fn store_feed_event_at( + pool: &PgPool, + community: CommunityId, + kind: u32, + content: &str, + channel_id: Option, + tags: Vec, + created_at: i64, + ) -> nostr::Event { + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(kind as u16), content) + .tags(tags) + .custom_created_at(nostr::Timestamp::from(created_at as u64)) + .sign_with_keys(&keys) + .expect("sign event"); + + let root_id: Option> = event.tags.iter().find_map(|tag| { + let t = tag.as_slice(); + (t.len() >= 4 && t[0] == "e" && t[3] == "root") + .then(|| hex::decode(&t[1]).expect("hex root id")) + }); + let event_created_at = + DateTime::from_timestamp(created_at, 0).expect("valid test timestamp"); + let thread_meta = match (&root_id, channel_id) { + (Some(root), Some(channel)) => Some(crate::event::ThreadMetadataParams { + event_id: event.id.as_bytes(), + event_created_at, + channel_id: channel, + parent_event_id: Some(root.as_slice()), + parent_event_created_at: None, + root_event_id: Some(root.as_slice()), + root_event_created_at: None, + depth: 1, + broadcast: false, + }), + _ => None, + }; + + crate::event::insert_event_with_thread_metadata( + pool, + community, + &event, + channel_id, + thread_meta, + ) + .await + .expect("insert feed event with thread metadata"); + crate::insert_mentions(pool, community, &event, channel_id) + .await + .expect("insert mentions"); + event + } + // -- Postgres tenant-scope regressions ------------------------------------ #[tokio::test] @@ -844,6 +963,134 @@ mod tests { ); } + #[test] + fn mentions_query_windows_per_conversation() { + let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); + let pubkey = vec![0x42; 32]; + let channel_id = Uuid::new_v4(); + let mut qb = build_mentions_query(community, &pubkey, &[channel_id], None, 10); + let query = qb.build(); + let sql_str = sqlx::Execute::sql(query); + let sql = sql_str.as_str(); + + assert!( + sql.contains("ROW_NUMBER() OVER (PARTITION BY k.conv_key"), + "mentions feed must rank events within each conversation: {sql}" + ); + assert!( + sql.contains(&format!("WHERE conv_rank <= {FEED_CONVERSATION_EVENT_CAP}")), + "mentions feed must cap events per conversation: {sql}" + ); + assert!( + sql.contains("'dm:' || c.channel_id::text"), + "DM conversations must key by channel, not thread root: {sql}" + ); + assert!( + sql.contains("COALESCE(tm.root_event_id, c.id)"), + "thread conversations must key by NIP-10 root with event-id fallback: {sql}" + ); + assert!( + sql.contains("ORDER BY conv_latest DESC"), + "conversations must surface newest-activity-first: {sql}" + ); + assert!( + sql.contains(&format!("LIMIT {FEED_WINDOW_SCAN_CAP}")), + "candidate scan must stay bounded: {sql}" + ); + } + + /// The regression this windowing exists to prevent: one high-volume + /// conversation must not starve every other conversation out of the + /// mentions window. + /// + /// Seeds one thread with 40 replies mentioning the user plus 6 standalone + /// mention events (6 distinct conversations), then asks for a 20-event + /// window. The old flat `ORDER BY created_at DESC LIMIT 20` returned the + /// 20 newest events — all from the chatty thread — so the 6 older + /// conversations vanished. The windowed query must return every + /// conversation, with the chatty one capped. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn query_mentions_survives_chatty_conversation_starvation() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let channel = insert_test_channel(&pool, community).await; + let mentioned_pubkey = "04".repeat(32); + let mentioned_bytes = hex::decode(&mentioned_pubkey).expect("hex pubkey"); + + let base = chrono::Utc::now().timestamp() - 10_000; + + // 6 standalone conversations, oldest first. + let mut standalone_ids = Vec::new(); + for n in 0..6 { + let event = store_feed_event_at( + &pool, + community, + KIND_STREAM_MESSAGE, + &format!("standalone {n}"), + Some(channel), + vec![Tag::parse(["p", mentioned_pubkey.as_str()]).unwrap()], + base + n * 10, + ) + .await; + standalone_ids.push(event.id); + } + + // One chatty thread: root + 40 newer replies, all mentioning the user. + let root = store_feed_event_at( + &pool, + community, + KIND_STREAM_MESSAGE, + "chatty root", + Some(channel), + vec![Tag::parse(["p", mentioned_pubkey.as_str()]).unwrap()], + base + 100, + ) + .await; + let root_hex = root.id.to_hex(); + for n in 0..40 { + store_feed_event_at( + &pool, + community, + KIND_STREAM_MESSAGE, + &format!("chatty reply {n}"), + Some(channel), + vec![ + Tag::parse(["p", mentioned_pubkey.as_str()]).unwrap(), + Tag::parse(["e", root_hex.as_str(), "", "root"]).unwrap(), + ], + base + 200 + n, + ) + .await; + } + + let rows = query_mentions(&pool, community, &mentioned_bytes, &[channel], None, 20) + .await + .expect("query windowed mentions"); + + assert!(rows.len() <= 20, "limit must hold: got {} rows", rows.len()); + for standalone in &standalone_ids { + assert!( + rows.iter().any(|row| row.event.id == *standalone), + "standalone conversation {standalone} must survive the chatty thread" + ); + } + let chatty_rows = rows + .iter() + .filter(|row| { + row.event.id == root.id + || row.event.tags.iter().any(|tag| { + let t = tag.as_slice(); + t.len() >= 2 && t[0] == "e" && t[1] == root_hex + }) + }) + .count(); + assert!( + chatty_rows as i64 <= FEED_CONVERSATION_EVENT_CAP, + "chatty conversation must be capped at {FEED_CONVERSATION_EVENT_CAP}, got {chatty_rows}" + ); + } + #[test] fn needs_action_query_is_tenant_scoped_and_joins_mentions_by_composite_key() { let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index e930f0ef612..d1f57c2298f 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -158,6 +158,7 @@ export default defineConfig({ "**/stream.spec.ts", "**/integration.spec.ts", "**/dm-double-notification.spec.ts", + "**/inbox-windowing-screenshots.spec.ts", "**/profile.spec.ts", "**/sidebar.spec.ts", "**/sidebar-relay-card.spec.ts", diff --git a/desktop/src-tauri/src/commands/messages.rs b/desktop/src-tauri/src/commands/messages.rs index 4f839638b93..38fd270afec 100644 --- a/desktop/src-tauri/src/commands/messages.rs +++ b/desktop/src-tauri/src/commands/messages.rs @@ -69,7 +69,15 @@ pub async fn get_feed( }; // Mentions: messages that reference me via #p. + // + // `feed_types` routes this filter through the relay's feed path, whose + // mentions query windows **per conversation** (newest conversations first, + // a few newest events each) instead of a flat newest-N-events cut. The + // flat cut let one chatty thread/DM fill the whole window and starve + // every other conversation out of the Inbox. The kinds below still bound + // the generic-query fallback for older relays that ignore `feed_types`. let mut mention_filter = serde_json::json!({ + "feed_types": ["mentions"], "kinds": [ 9, 40002, diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 4a224709ea0..65decf04c58 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -7366,9 +7366,16 @@ async function handleGetFeed( // Feed is composed of multiple queries: mentions (#p), activity, approvals. // For e2e, return a minimal feed structure with mentions. + // + // `feed_types` mirrors the native bridge (commands/messages.rs get_feed): + // it routes the filter through the relay's per-conversation-windowed feed + // path instead of the flat newest-N generic query. Keeping the e2e bridge + // faithful here is what lets the inbox windowing e2e exercise the same + // relay path the desktop app uses. const limit = args.limit ?? 50; const mentionEvents = await relayQuery(config, [ { + feed_types: ["mentions"], kinds: [ 9, 40002, diff --git a/desktop/tests/e2e/inbox-windowing-screenshots.spec.ts b/desktop/tests/e2e/inbox-windowing-screenshots.spec.ts new file mode 100644 index 00000000000..156ea3ba22d --- /dev/null +++ b/desktop/tests/e2e/inbox-windowing-screenshots.spec.ts @@ -0,0 +1,207 @@ +/** + * Inbox feed windowing — starvation regression + before/after screenshots. + * + * The mentions feed used to be a flat "newest N events p-tagging me" window. + * Clients group events into conversation rows *after* that cut, so one chatty + * DM or thread could occupy every slot and starve all other conversations out + * of the Inbox entirely. The relay now windows the mentions feed + * per-conversation (crates/buzz-db/src/feed.rs build_mentions_query) when the + * query carries `feed_types`. + * + * Both tests seed the same shape: one chatty DM (60 messages, newest) plus 6 + * standalone mentions in #general (older). The "before" test strips + * `feed_types` from the outgoing /query — reproducing the flat generic-query + * window production used before this change — and shows the starved inbox. + * The "after" test uses the windowed feed path and shows every conversation + * surviving. + * + * Run (against an isolated relay): + * BUZZ_E2E_RELAY_URL=http://localhost:3030 pnpm build:e2e && \ + * pnpm exec playwright test --project=integration \ + * tests/e2e/inbox-windowing-screenshots.spec.ts + * Output: test-results/inbox-windowing/ + */ +import { expect, test, type Page, type Route } from "@playwright/test"; +import { hexToBytes } from "@noble/hashes/utils.js"; +import { finalizeEvent, type VerifiedEvent } from "nostr-tools/pure"; + +import { installRelayBridge, TEST_IDENTITIES } from "../helpers/bridge"; +import { assertRelaySeeded } from "../helpers/seed"; + +const SHOTS = "test-results/inbox-windowing"; +const RELAY_HTTP = process.env.BUZZ_E2E_RELAY_URL ?? "http://localhost:3000"; + +// uuid5("buzz.channel.dm.alice-tyler") / uuid5("buzz.channel.general") — +// fixed ids seeded by scripts/setup-desktop-test-data.sh. +const DM_ID = "5a9c064e-0411-5242-ae6b-0363ba99b8e6"; +const GENERAL_ID = "9f28288a-d724-587a-9709-92dc7f967110"; + +const CHATTY_COUNT = 60; +const STANDALONE_COUNT = 6; + +// Unique per run so reruns against a shared relay stay deterministic: the +// assertions only track this run's markers, never leftovers. +const RUN_ID = Math.random().toString(36).slice(2, 8); + +async function publish( + sender: keyof typeof TEST_IDENTITIES, + channelId: string, + content: string, + createdAt: number, +): Promise { + const event = finalizeEvent( + { + kind: 9, + content, + created_at: createdAt, + tags: [ + ["h", channelId], + ["p", TEST_IDENTITIES.tyler.pubkey], + ], + }, + hexToBytes(TEST_IDENTITIES[sender].privateKey), + ); + const response = await fetch(`${RELAY_HTTP}/events`, { + method: "POST", + headers: { "Content-Type": "application/json", "X-Pubkey": event.pubkey }, + body: JSON.stringify(event), + }); + if (!response.ok) { + throw new Error( + `POST /events failed (${response.status}): ${await response.text()}`, + ); + } + return event; +} + +function standaloneMarker(index: number) { + return `standalone mention ${index} [${RUN_ID}]`; +} + +/** + * Seed the starvation shape: 6 standalone mentions in #general (older), + * then a 60-message chatty DM burst (newest). A flat 50-event window is + * entirely consumed by the burst. + * + * The relay rejects events drifting more than ±15 min from server time + * (ingest.rs MAX_TIMESTAMP_DRIFT_SECS), so the whole timeline is compressed + * into the last ~14 minutes: mentions at −840s spaced 10s, burst at −600s + * spaced 5s. Ordering is what matters, not the span. + */ +async function seedStarvationShape() { + const nowSecond = Math.floor(Date.now() / 1000); + for (let i = 0; i < STANDALONE_COUNT; i++) { + await publish( + "bob", + GENERAL_ID, + standaloneMarker(i), + nowSecond - 840 + i * 10, + ); + } + for (let i = 0; i < CHATTY_COUNT; i++) { + await publish( + "alice", + DM_ID, + `chatty dm update ${i} [${RUN_ID}]`, + nowSecond - 600 + i * 5, + ); + } +} + +/** + * Strip `feed_types` from outgoing /query filters — the pre-windowing wire + * shape. The relay then serves the mention filter through the flat + * generic-query path, which is exactly what production clients sent before + * this change. + */ +async function forceFlatFeedWindow(page: Page) { + await page.route( + (url) => url.pathname.endsWith("/query"), + async (route: Route) => { + const request = route.request(); + let body: unknown; + try { + body = request.postDataJSON(); + } catch { + await route.continue(); + return; + } + if (!Array.isArray(body)) { + await route.continue(); + return; + } + const stripped = body.map((filter) => { + if (filter && typeof filter === "object" && "feed_types" in filter) { + const { feed_types: _dropped, ...rest } = filter as Record< + string, + unknown + >; + return rest; + } + return filter; + }); + await route.continue({ postData: JSON.stringify(stripped) }); + }, + ); +} + +function getListPane(page: Page) { + return page.getByTestId("home-inbox-list"); +} + +test.describe("inbox feed windowing", () => { + test.use({ viewport: { width: 1280, height: 900 } }); + + test.beforeAll(async () => { + test.setTimeout(120_000); + await assertRelaySeeded(); + await seedStarvationShape(); + }); + + test("before — flat window lets one chatty DM starve the inbox", async ({ + page, + }) => { + await forceFlatFeedWindow(page); + await installRelayBridge(page, "tyler"); + await page.goto("/"); + const list = getListPane(page); + await expect(list).toBeVisible(); + + // The chatty DM survives as a single collapsed row… + await expect(list).toContainText(`chatty dm update ${CHATTY_COUNT - 1}`, { + timeout: 15_000, + }); + // …but every standalone mention fell off the flat 50-event cliff. + for (let i = 0; i < STANDALONE_COUNT; i++) { + await expect(list).not.toContainText(standaloneMarker(i)); + } + + await page.screenshot({ + path: `${SHOTS}/01-before-flat-window-starved.png`, + fullPage: true, + }); + }); + + test("after — per-conversation windowing keeps every conversation", async ({ + page, + }) => { + await installRelayBridge(page, "tyler"); + await page.goto("/"); + const list = getListPane(page); + await expect(list).toBeVisible(); + + // The chatty DM still shows (capped, collapsed to one row)… + await expect(list).toContainText(`chatty dm update ${CHATTY_COUNT - 1}`, { + timeout: 15_000, + }); + // …and every standalone mention conversation survives the window. + for (let i = 0; i < STANDALONE_COUNT; i++) { + await expect(list).toContainText(standaloneMarker(i)); + } + + await page.screenshot({ + path: `${SHOTS}/02-after-windowed-conversations.png`, + fullPage: true, + }); + }); +});