From a6955dfef2069f1d210c9392783ce8c2d5c60765 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Mon, 24 Aug 2026 20:11:20 +0100 Subject: [PATCH 1/6] Resurface hidden DMs on new messages Signed-off-by: kenny lopez --- crates/buzz-core/src/kind.rs | 27 ++++ crates/buzz-db/src/dm.rs | 39 ++++++ crates/buzz-db/src/lib.rs | 13 ++ crates/buzz-relay/src/handlers/ingest.rs | 68 ++++++---- .../buzz-relay/src/handlers/side_effects.rs | 31 +++++ .../tests/e2e_nostr_interop.rs | 111 +++++++++++++++- .../useMembershipNotifications.test.mjs | 123 ++++++++++++++++++ .../channels/useMembershipNotifications.ts | 69 +++++++--- .../features/channels/channels_provider.dart | 24 ++-- .../channels_provider/dm_visibility.dart | 81 ++++++++++++ .../channels/channels_provider_test.dart | 69 +++++++++- 11 files changed, 595 insertions(+), 60 deletions(-) create mode 100644 desktop/src/features/channels/useMembershipNotifications.test.mjs create mode 100644 mobile/lib/features/channels/channels_provider/dm_visibility.dart diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 3c6f1d5913d..89651507636 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -790,6 +790,17 @@ pub const fn is_workflow_execution_kind(kind: u32) -> bool { kind >= KIND_WORKFLOW_TRIGGERED && kind <= KIND_WORKFLOW_APPROVAL_DENIED } +/// Returns `true` for channel events that clients render as new message content. +/// +/// Reactions, edits, deletions, and system events are deliberately excluded: +/// they must not resurface a direct message without a new human-visible message. +pub const fn is_human_visible_message_kind(kind: u32) -> bool { + matches!( + kind, + KIND_STREAM_MESSAGE | KIND_STREAM_MESSAGE_V2 | KIND_FORUM_POST | KIND_FORUM_COMMENT + ) +} + /// Returns `true` if `kind` is a NIP-43 relay membership admin command (9030–9032) /// or the Buzz workspace-profile admin command (9033). pub const fn is_relay_admin_kind(kind: u32) -> bool { @@ -933,6 +944,22 @@ mod tests { } } + #[test] + fn human_visible_message_kind_matches_client_message_sets() { + for kind in [ + KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_V2, + KIND_FORUM_POST, + KIND_FORUM_COMMENT, + ] { + assert!(is_human_visible_message_kind(kind), "kind {kind}"); + } + + for kind in [KIND_REACTION, KIND_STREAM_MESSAGE_EDIT, KIND_DELETION] { + assert!(!is_human_visible_message_kind(kind), "kind {kind}"); + } + } + // ── event_is_shared / is_unshared_gated_event ──────────────────────── fn make_event_of_kind(kind: u32, tags: &[&[&str]]) -> nostr::Event { diff --git a/crates/buzz-db/src/dm.rs b/crates/buzz-db/src/dm.rs index 89e15c70260..e5bfe16fc6d 100644 --- a/crates/buzz-db/src/dm.rs +++ b/crates/buzz-db/src/dm.rs @@ -448,6 +448,45 @@ pub async fn unhide_dm( Ok(()) } +/// Clear the hidden state for every active recipient of a DM message. +/// +/// The sender is deliberately excluded so sending from another surface does +/// not rewrite their sidebar preference. Returns only viewers whose hidden +/// state changed, allowing the relay to publish targeted visibility snapshots. +pub async fn unhide_dm_recipients( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + sender_pubkey: &[u8], +) -> Result>> { + let rows = sqlx::query( + r#" + UPDATE channel_members cm + SET hidden_at = NULL + FROM channels c + WHERE cm.community_id = $1 + AND cm.channel_id = $2 + AND cm.pubkey != $3 + AND cm.removed_at IS NULL + AND cm.hidden_at IS NOT NULL + AND c.community_id = cm.community_id + AND c.id = cm.channel_id + AND c.channel_type = 'dm' + AND c.deleted_at IS NULL + RETURNING cm.pubkey + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(sender_pubkey) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| row.try_get::, _>("pubkey").map_err(Into::into)) + .collect() +} + /// Return the channel IDs of all DMs the given user currently has hidden /// (`hidden_at IS NOT NULL`) while still being an active member. Used to build /// the relay-signed NIP-DV visibility snapshot. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3ff230f9503..ee092b3c85f 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2897,6 +2897,19 @@ impl Db { dm::unhide_dm(&self.pool, community_id, channel_id, pubkey).await } + /// Unhide a DM for active recipients other than the message sender. + /// + /// Returns only viewers whose hidden state changed. + #[datastore_span(name = "unhide_dm_recipients", system = "postgresql")] + pub async fn unhide_dm_recipients( + &self, + community_id: CommunityId, + channel_id: Uuid, + sender_pubkey: &[u8], + ) -> Result>> { + dm::unhide_dm_recipients(&self.pool, community_id, channel_id, sender_pubkey).await + } + /// List the channel IDs of all DMs the given user currently has hidden. #[datastore_span(name = "list_hidden_dms", system = "postgresql")] pub async fn list_hidden_dms( diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index dd2fa6e93e0..50b87ee0f95 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -11,30 +11,30 @@ use uuid::Uuid; use buzz_auth::Scope; use buzz_core::kind::{ - event_kind_u32, is_identity_archive_request_kind, is_parameterized_replaceable, - is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, - KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_AUTH, KIND_BOOKMARK_LIST, KIND_BOOKMARK_SET, - KIND_CANVAS, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, KIND_DM_HIDE, KIND_DM_OPEN, - KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, KIND_FOLLOW_SET, KIND_FORUM_COMMENT, - KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, KIND_GIT_ISSUE, KIND_GIT_PATCH, - KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, - KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, - KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, KIND_HUDDLE_PARTICIPANT_JOINED, - KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, KIND_IA_ARCHIVE_REQUEST, - KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, KIND_MEMBER_ADDED_NOTIFICATION, - KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, - KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, - KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, KIND_NIP29_DELETE_GROUP, - KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, - KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_NIP43_LEAVE_REQUEST, - KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, KIND_PRESENCE_UPDATE, - KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_PROJECT, KIND_REACTION, - KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, - KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, - KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, - KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, - RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, - RELAY_ADMIN_SET_WORKSPACE_PROFILE, + event_kind_u32, is_human_visible_message_kind, is_identity_archive_request_kind, + is_parameterized_replaceable, is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, + KIND_AGENT_TURN_METRIC, KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_AUTH, KIND_BOOKMARK_LIST, + KIND_BOOKMARK_SET, KIND_CANVAS, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, + KIND_DM_HIDE, KIND_DM_OPEN, KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, + KIND_FOLLOW_SET, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, + KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, + KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, + KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, + KIND_HUDDLE_PARTICIPANT_JOINED, KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, + KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, + KIND_MODERATION_RESOLVE_REPORT, KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, + KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, + KIND_NIP29_DELETE_GROUP, KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, + KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, + KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, + KIND_PRESENCE_UPDATE, KIND_PRIVATE_MANAGED_AGENT, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, + KIND_PROJECT, KIND_REACTION, KIND_READ_STATE, KIND_REPORT, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, + KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, + KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEAM_CATALOG, KIND_TEXT_NOTE, KIND_USER_STATUS, + KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, + RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; @@ -3197,6 +3197,26 @@ async fn ingest_event_inner( }); } + if is_human_visible_message_kind(kind_u32) { + if let (Some(ch_id), Some(channel)) = (channel_id, channel_row.as_ref()) { + if channel.channel_type == "dm" { + let sender = effective_message_author(&event, &state.relay_keypair.public_key()); + if let Err(error) = + crate::handlers::side_effects::resurface_dm_for_message_recipients( + tenant, state, ch_id, &sender, + ) + .await + { + error!( + event_id = %event_id_hex, + channel_id = %ch_id, + "Failed to resurface DM for message recipients: {error}" + ); + } + } + } + } + if crate::handlers::side_effects::is_side_effect_kind(kind_u32) { if let Err(e) = crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state) diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 89595fbee17..b51dd79016a 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3469,6 +3469,37 @@ pub async fn publish_dm_visibility_snapshot( Ok(()) } +/// Resurface a DM for recipients of a newly accepted message. +/// +/// Hidden state is per viewer, so only active members other than the effective +/// author are changed. A fresh snapshot makes the update visible across clients. +pub async fn resurface_dm_for_message_recipients( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + sender_pubkey: &[u8], +) -> anyhow::Result<()> { + let recipients = state + .db + .unhide_dm_recipients(tenant.community(), channel_id, sender_pubkey) + .await?; + + let mut failures = Vec::new(); + for recipient in recipients { + if let Err(error) = publish_dm_visibility_snapshot(tenant, state, &recipient).await { + failures.push(error.to_string()); + } + } + if !failures.is_empty() { + anyhow::bail!( + "failed to publish DM visibility snapshots: {}", + failures.join("; ") + ); + } + + Ok(()) +} + #[allow(clippy::too_many_arguments)] async fn publish_nipia_delta( tenant: &TenantContext, diff --git a/crates/buzz-test-client/tests/e2e_nostr_interop.rs b/crates/buzz-test-client/tests/e2e_nostr_interop.rs index fce78776764..8c1c49b6b2f 100644 --- a/crates/buzz-test-client/tests/e2e_nostr_interop.rs +++ b/crates/buzz-test-client/tests/e2e_nostr_interop.rs @@ -80,11 +80,16 @@ async fn create_test_channel(keys: &Keys) -> String { channel_uuid.to_string() } -/// Send a message via a signed kind:9 event and return the event_id hex. -async fn send_rest_message(keys: &Keys, channel_id: &str, content: &str) -> String { +/// Send a message with an explicit supported kind and return the event id. +async fn send_rest_message_with_kind( + keys: &Keys, + channel_id: &str, + content: &str, + kind: u16, +) -> String { let client = reqwest::Client::new(); let pubkey_hex = keys.public_key().to_hex(); - let event = EventBuilder::new(Kind::Custom(9), content) + let event = EventBuilder::new(Kind::Custom(kind), content) .tags(vec![Tag::parse(["h", channel_id]).unwrap()]) .sign_with_keys(keys) .unwrap(); @@ -105,17 +110,25 @@ async fn send_rest_message(keys: &Keys, channel_id: &str, content: &str) -> Stri body["event_id"].as_str().expect("event_id").to_string() } +async fn send_rest_message(keys: &Keys, channel_id: &str, content: &str) -> String { + send_rest_message_with_kind(keys, channel_id, content, 9).await +} + /// Create a DM via a signed kind:41010 (DM open) command event and return the /// channel_id UUID string parsed from the relay's `response:{...}` message. -async fn create_dm(requester_keys: &Keys, other_pubkey_hex: &str) -> String { +async fn create_group_dm(requester_keys: &Keys, other_pubkey_hexes: &[String]) -> String { let client = reqwest::Client::new(); let pubkey_hex = requester_keys.public_key().to_hex(); // Backdate the initial open so a later re-open kind:41010 with identical // tags in the same wall-clock second does not produce an identical event id // (which the relay would dedupe as "duplicate: already processed"). let backdated = nostr::Timestamp::from(nostr::Timestamp::now().as_secs() - 10); + let tags = other_pubkey_hexes + .iter() + .map(|pubkey| Tag::parse(["p", pubkey]).unwrap()) + .collect::>(); let event = EventBuilder::new(Kind::Custom(41010), "") - .tags(vec![Tag::parse(["p", other_pubkey_hex]).unwrap()]) + .tags(tags) .custom_created_at(backdated) .sign_with_keys(requester_keys) .unwrap(); @@ -146,6 +159,10 @@ async fn create_dm(requester_keys: &Keys, other_pubkey_hex: &str) -> String { .to_string() } +async fn create_dm(requester_keys: &Keys, other_pubkey_hex: &str) -> String { + create_group_dm(requester_keys, &[other_pubkey_hex.to_string()]).await +} + /// Submit a signed command event via REST and assert it was accepted. async fn post_signed_event(keys: &Keys, kind: u16, tags: Vec) { let client = reqwest::Client::new(); @@ -1309,6 +1326,90 @@ async fn test_nipdv_hide_then_reopen_updates_snapshot() { client_a.disconnect().await.expect("disconnect"); } +/// Every accepted human-visible message kind resurfaces a hidden DM for the +/// recipient while preserving the sender and unrelated conversations. +#[tokio::test] +#[ignore] +async fn test_nipdv_supported_message_kinds_resurface_hidden_dm_for_recipient() { + for kind in [9u16, 40002, 45001, 45003] { + let keys_a = Keys::generate(); + let keys_b = Keys::generate(); + let keys_c = Keys::generate(); + let a_pubkey_hex = keys_a.public_key().to_hex(); + let b_pubkey_hex = keys_b.public_key().to_hex(); + let c_pubkey_hex = keys_c.public_key().to_hex(); + let channel_id = create_dm(&keys_a, &b_pubkey_hex).await; + let unrelated_channel_id = create_dm(&keys_b, &c_pubkey_hex).await; + + for (keys, dm_id) in [ + (&keys_a, channel_id.as_str()), + (&keys_b, channel_id.as_str()), + (&keys_b, unrelated_channel_id.as_str()), + ] { + post_signed_event(keys, 41012, vec![Tag::parse(["h", dm_id]).unwrap()]).await; + } + + send_rest_message_with_kind(&keys_a, &channel_id, "new inbound activity", kind).await; + + let mut client_a = BuzzTestClient::connect(&relay_url(), &keys_a) + .await + .expect("client A connect"); + let a_hidden = read_hidden_dms(&mut client_a, &a_pubkey_hex).await; + assert!( + a_hidden.contains(&channel_id), + "kind {kind} must preserve the sender's hidden state; A sees: {a_hidden:?}" + ); + client_a.disconnect().await.expect("disconnect A"); + + let mut client_b = BuzzTestClient::connect(&relay_url(), &keys_b) + .await + .expect("client B connect"); + let b_hidden = read_hidden_dms(&mut client_b, &b_pubkey_hex).await; + assert!( + !b_hidden.contains(&channel_id), + "kind {kind} must resurface the DM for B; B sees: {b_hidden:?}" + ); + assert!( + b_hidden.contains(&unrelated_channel_id), + "kind {kind} must preserve unrelated hidden DMs; B sees: {b_hidden:?}" + ); + client_b.disconnect().await.expect("disconnect B"); + } +} + +/// Group-DM delivery resurfaces every hidden recipient, not only the first. +#[tokio::test] +#[ignore] +async fn test_nipdv_group_message_resurfaces_all_hidden_recipients() { + let keys_a = Keys::generate(); + let keys_b = Keys::generate(); + let keys_c = Keys::generate(); + let b_pubkey_hex = keys_b.public_key().to_hex(); + let c_pubkey_hex = keys_c.public_key().to_hex(); + let channel_id = create_group_dm(&keys_a, &[b_pubkey_hex.clone(), c_pubkey_hex.clone()]).await; + + for keys in [&keys_b, &keys_c] { + post_signed_event(keys, 41012, vec![Tag::parse(["h", &channel_id]).unwrap()]).await; + } + + send_rest_message(&keys_a, &channel_id, "group activity").await; + + for (label, keys, pubkey) in [("B", &keys_b, &b_pubkey_hex), ("C", &keys_c, &c_pubkey_hex)] { + let mut client = BuzzTestClient::connect(&relay_url(), keys) + .await + .unwrap_or_else(|error| panic!("client {label} connect: {error}")); + let hidden = read_hidden_dms(&mut client, pubkey).await; + assert!( + !hidden.contains(&channel_id), + "new group activity must resurface the DM for {label}; hidden: {hidden:?}" + ); + client + .disconnect() + .await + .unwrap_or_else(|error| panic!("disconnect {label}: {error}")); + } +} + /// NIP-DV monotonicity regression: a hide immediately followed by a re-open /// within the same wall-clock second must still leave the re-open authoritative. /// diff --git a/desktop/src/features/channels/useMembershipNotifications.test.mjs b/desktop/src/features/channels/useMembershipNotifications.test.mjs new file mode 100644 index 00000000000..8ebfa550132 --- /dev/null +++ b/desktop/src/features/channels/useMembershipNotifications.test.mjs @@ -0,0 +1,123 @@ +import assert from "node:assert/strict"; +import { after, before, test } from "node:test"; + +import { JSDOM } from "jsdom"; + +const dom = new JSDOM("", { + url: "http://localhost", +}); + +before(() => { + Object.assign(globalThis, { + document: dom.window.document, + HTMLElement: dom.window.HTMLElement, + IS_REACT_ACT_ENVIRONMENT: true, + window: dom.window, + }); +}); + +after(() => dom.window.close()); + +test("DM visibility has a dedicated replay and refreshes only the channel list", async () => { + const React = await import("react"); + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { QueryClient, QueryClientProvider } = await import( + "@tanstack/react-query" + ); + const { relayClient } = await import("@/shared/api/relayClient"); + const { + KIND_DM_VISIBILITY, + KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, + } = await import("@/shared/constants/kinds"); + const { useMembershipNotifications } = await import( + "./useMembershipNotifications.ts" + ); + + const originalSubscribeLive = relayClient.subscribeLive; + const subscriptions = []; + relayClient.subscribeLive = async (filter, listener) => { + subscriptions.push({ filter, listener }); + return async () => {}; + }; + + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const invalidations = []; + queryClient.invalidateQueries = async ({ queryKey }) => { + invalidations.push(queryKey); + }; + const wrapper = ({ children }) => + React.createElement(QueryClientProvider, { client: queryClient }, children); + + try { + const { unmount } = renderHook( + () => useMembershipNotifications("Viewer-Pubkey"), + { wrapper }, + ); + await act(async () => new Promise((resolve) => setImmediate(resolve))); + + assert.equal(subscriptions.length, 2); + const membership = subscriptions.find((subscription) => + subscription.filter.kinds.includes(KIND_MEMBER_ADDED_NOTIFICATION), + ); + const visibility = subscriptions.find( + (subscription) => + subscription.filter.kinds.length === 1 && + subscription.filter.kinds[0] === KIND_DM_VISIBILITY, + ); + assert.deepEqual(membership.filter.kinds, [ + KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, + ]); + assert.deepEqual(membership.filter["#p"], ["viewer-pubkey"]); + assert.equal(membership.filter.limit, 50); + assert.deepEqual(visibility.filter["#p"], ["viewer-pubkey"]); + assert.equal(visibility.filter.limit, 1); + assert.equal(visibility.filter.since, undefined); + + await act(async () => { + visibility.listener({ + id: "visibility", + pubkey: "relay", + created_at: 1, + kind: KIND_DM_VISIBILITY, + tags: [ + ["p", "viewer-pubkey"], + ["h", "still-hidden-dm"], + ], + content: "", + sig: "sig", + }); + }); + assert.deepEqual(invalidations, [["channels"]]); + + invalidations.length = 0; + await act(async () => { + membership.listener({ + id: "membership", + pubkey: "relay", + created_at: 2, + kind: KIND_MEMBER_ADDED_NOTIFICATION, + tags: [ + ["p", "viewer-pubkey"], + ["h", "new-channel"], + ], + content: "", + sig: "sig", + }); + }); + assert.deepEqual(invalidations, [ + ["channels"], + ["channels", "new-channel", "detail"], + ["channels", "new-channel", "members"], + ]); + + unmount(); + } finally { + cleanup(); + queryClient.clear(); + relayClient.subscribeLive = originalSubscribeLive; + } +}); diff --git a/desktop/src/features/channels/useMembershipNotifications.ts b/desktop/src/features/channels/useMembershipNotifications.ts index 7f8fa91a6a6..f66e672b71c 100644 --- a/desktop/src/features/channels/useMembershipNotifications.ts +++ b/desktop/src/features/channels/useMembershipNotifications.ts @@ -6,6 +6,7 @@ import { getChannelIdFromTags } from "@/features/messages/lib/threading"; import { relayClient } from "@/shared/api/relayClient"; import type { RelayEvent } from "@/shared/api/types"; import { + KIND_DM_VISIBILITY, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, } from "@/shared/constants/kinds"; @@ -22,6 +23,9 @@ export function useMembershipNotifications(currentPubkey?: string) { const channelId = getChannelIdFromTags(event.tags); void queryClient.invalidateQueries({ queryKey: channelsQueryKey }); + if (event.kind === KIND_DM_VISIBILITY) { + return; + } if (!channelId) { return; } @@ -46,31 +50,62 @@ export function useMembershipNotifications(currentPubkey?: string) { let dispose: (() => Promise) | undefined; const subscribe = async (): Promise => { + const nextDisposes: Array<() => Promise> = []; try { - const nextDispose = await relayClient.subscribeLive( - { - kinds: [ - KIND_MEMBER_ADDED_NOTIFICATION, - KIND_MEMBER_REMOVED_NOTIFICATION, - ], - "#p": [normalizedCurrentPubkey], - limit: 50, - since: Math.floor(Date.now() / 1_000) - 30, - }, - (event) => { - if (!isCancelled) { - handleMembershipNotification(event); - } - }, + const handleEvent = (event: RelayEvent) => { + if (!isCancelled) { + handleMembershipNotification(event); + } + }; + nextDisposes.push( + await relayClient.subscribeLive( + { + kinds: [ + KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, + ], + "#p": [normalizedCurrentPubkey], + limit: 50, + since: Math.floor(Date.now() / 1_000) - 30, + }, + handleEvent, + ), + ); + nextDisposes.push( + await relayClient.subscribeLive( + { + kinds: [KIND_DM_VISIBILITY], + "#p": [normalizedCurrentPubkey], + // A separate one-event replay closes the history/subscription + // gap without sharing membership notifications' replay budget. + limit: 1, + }, + handleEvent, + ), ); + const nextDispose = async () => { + await Promise.all( + nextDisposes.map((unsubscribe) => + unsubscribe().catch(() => undefined), + ), + ); + }; if (isCancelled) { - void nextDispose().catch(() => {}); + void nextDispose(); return true; } dispose = nextDispose; return true; } catch (error) { - console.error("Failed to subscribe to membership notifications", error); + await Promise.all( + nextDisposes.map((unsubscribe) => + unsubscribe().catch(() => undefined), + ), + ); + console.error( + "Failed to subscribe to channel-list notifications", + error, + ); return false; } }; diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 2258a493c99..e7135dc2531 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -20,6 +20,7 @@ import 'unread_badge/observed_unread_event.dart'; import 'unread_badge/should_notify_for_event.dart'; part 'channel_directory.dart'; +part 'channels_provider/dm_visibility.dart'; const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2}; const _unreadCatchUpLimit = 1000; @@ -36,12 +37,14 @@ const _authoredRootIdsPrefix = 'buzz-thread-authored.v1'; /// Live updates are layered on top via per-channel subscriptions on the /// `#h` tag for any of the visible channel event kinds — incoming events /// bump `lastMessageAt` for that channel. -class ChannelsNotifier extends AsyncNotifier> { +class ChannelsNotifier extends AsyncNotifier> + with _DmVisibilitySubscription { static const _backstopInterval = Duration(seconds: 60); final Map _unsubscribersByChannel = {}; Future _liveSubscriptionQueue = Future.value(); Set _desiredLiveChannelIds = const {}; + @override int _subscriptionVersion = 0; String? _subscriptionRelayBaseUrl; Timer? _backstopTimer; @@ -590,6 +593,15 @@ class ChannelsNotifier extends AsyncNotifier> { final session = ref.read(relaySessionProvider.notifier); final channelIds = _desiredLiveChannelIds; + final myPk = ref.read(myPubkeyProvider)?.toLowerCase(); + await _syncDmVisibilitySubscription( + relayBaseUrl, + subscriptionVersion, + fence, + session, + myPk, + ); + for (final entry in _unsubscribersByChannel.entries.toList()) { if (channelIds.contains(entry.key)) continue; _unsubscribersByChannel.remove(entry.key); @@ -814,14 +826,6 @@ class ChannelsNotifier extends AsyncNotifier> { }); } - Set _mutedChannelIds() => { - for (final entry in ref.read(channelMutesProvider).store.channels.entries) - if (entry.value.muted) entry.key, - }; - - Set _followedRootIds() => - ref.read(threadFollowsProvider).followedRootIds; - void _loadThreadInterestStores(String pubkey) { final normalizedPubkey = pubkey.toLowerCase(); if (_threadInterestPubkey == normalizedPubkey) return; @@ -906,6 +910,7 @@ class ChannelsNotifier extends AsyncNotifier> { } } + @override Future refresh() async { final sessionState = ref.read(relaySessionProvider); // Don't attempt to fetch when the session isn't connected — fetchHistory @@ -983,6 +988,7 @@ class ChannelsNotifier extends AsyncNotifier> { } _unsubscribersByChannel.clear(); _subscriptionRelayBaseUrl = null; + _clearDmVisibilitySubscription(); _backstopTimer?.cancel(); _backstopTimer = null; } diff --git a/mobile/lib/features/channels/channels_provider/dm_visibility.dart b/mobile/lib/features/channels/channels_provider/dm_visibility.dart new file mode 100644 index 00000000000..1820027666e --- /dev/null +++ b/mobile/lib/features/channels/channels_provider/dm_visibility.dart @@ -0,0 +1,81 @@ +part of '../channels_provider.dart'; + +mixin _DmVisibilitySubscription on AsyncNotifier> { + void Function()? _unsubscribeDmVisibility; + String? _dmVisibilityRelayBaseUrl; + String? _dmVisibilityPubkey; + + int get _subscriptionVersion; + + Future refresh(); + + Future _syncDmVisibilitySubscription( + String relayBaseUrl, + int subscriptionVersion, + _ChannelRefreshFence fence, + RelaySessionNotifier session, + String? myPk, + ) async { + if (_dmVisibilityRelayBaseUrl != relayBaseUrl || + _dmVisibilityPubkey != myPk) { + _unsubscribeDmVisibility?.call(); + _unsubscribeDmVisibility = null; + _dmVisibilityRelayBaseUrl = relayBaseUrl; + _dmVisibilityPubkey = myPk; + } + if (_unsubscribeDmVisibility != null || myPk == null) return; + + try { + final unsubscribe = await session.subscribe( + NostrFilter( + kinds: const [EventKind.dmVisibility], + tags: { + '#p': [myPk], + }, + // Replay the latest snapshot so publication in the gap after the + // history fetch still refreshes the channel list. + limit: 1, + ), + (event) { + if (ref.read(relayConfigProvider).baseUrl == relayBaseUrl && + ref.read(myPubkeyProvider)?.toLowerCase() == myPk) { + _handleDmVisibilityEvent(event); + } + }, + ); + if (!fence.isCurrent || + subscriptionVersion != _subscriptionVersion || + ref.read(relaySessionProvider).status != SessionStatus.connected || + ref.read(relayConfigProvider).baseUrl != relayBaseUrl || + ref.read(myPubkeyProvider)?.toLowerCase() != myPk) { + unsubscribe(); + return; + } + _unsubscribeDmVisibility = unsubscribe; + } catch (error) { + debugPrint( + '[ChannelsNotifier] DM visibility subscription failed: $error', + ); + } + } + + void _handleDmVisibilityEvent(NostrEvent event) { + if (event.kind != EventKind.dmVisibility) return; + unawaited(refresh()); + } + + void _clearDmVisibilitySubscription() { + _unsubscribeDmVisibility?.call(); + _unsubscribeDmVisibility = null; + _dmVisibilityRelayBaseUrl = null; + _dmVisibilityPubkey = null; + } + + Set _mutedChannelIds() => { + for (final entry in ref.read(channelMutesProvider).store.channels.entries) + if (entry.value.muted) entry.key, + }; + + Set _followedRootIds() => + ref.read(threadFollowsProvider).followedRootIds; +} diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index ea6c38cdfd1..de5a310be91 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -2054,6 +2054,55 @@ void main() { ), isTrue, ); + expect(session.visibilitySubscribeFilters, hasLength(1)); + expect(session.visibilitySubscribeFilters.single.limit, 1); + expect(session.visibilitySubscribeFilters.single.tags['#p'], [myPk]); + }); + + test('a DM visibility snapshot refreshes the hidden channel list', () async { + final hiddenDmEvents = [ + _hiddenDms([_channelA], pubkey: myPk), + ]; + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk), _membership(_channelB, myPk)], + metadata: [ + _meta(id: _channelA, name: 'Alice', channelType: 'dm'), + _meta(id: _channelB, name: 'Bob', channelType: 'dm'), + ], + hiddenDmEvents: hiddenDmEvents, + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + + expect( + (await container.read( + channelsProvider.future, + )).map((channel) => channel.id), + [_channelB], + ); + + hiddenDmEvents + ..clear() + ..add(_hiddenDms(const [], pubkey: myPk)); + session.emit(hiddenDmEvents.single); + + await _waitUntil( + () => + container + .read(channelsProvider) + .value + ?.map((channel) => channel.id) + .toSet() + .length == + 2, + ); + expect( + container + .read(channelsProvider) + .requireValue + .map((channel) => channel.id), + unorderedEquals([_channelA, _channelB]), + ); }); test( @@ -2424,6 +2473,7 @@ class _FakeRelaySession extends RelaySessionNotifier { final List directoryQueryFilters = []; final List membershipQueryFilters = []; final List subscribeFilters = []; + final List visibilitySubscribeFilters = []; final Map _subscriptions = {}; int _nextSubscriptionKey = 0; Completer? _pausedSubscribe; @@ -2449,7 +2499,7 @@ class _FakeRelaySession extends RelaySessionNotifier { for (final (filter, _) in _subscriptions.values) ?filter.tags['#h']?.single, }; - int get activeSubscriptionCount => _subscriptions.length; + int get activeSubscriptionCount => activeChannels.length; Future get nextSubscribeStarted async { final started = _subscribeStarted; @@ -2815,8 +2865,13 @@ class _FakeRelaySession extends RelaySessionNotifier { void Function(NostrEvent) onEvent, { void Function(String message)? onClosed, }) async { - totalSubscribeCount++; - subscribeFilters.add(filter); + final isChannelSubscription = filter.tags['#h'] != null; + if (isChannelSubscription) { + totalSubscribeCount++; + subscribeFilters.add(filter); + } else if (filter.kinds.contains(EventKind.dmVisibility)) { + visibilitySubscribeFilters.add(filter); + } final paused = _pausedSubscribe; if (paused != null) { _subscribeStarted!.complete(); @@ -2829,8 +2884,12 @@ class _FakeRelaySession extends RelaySessionNotifier { return () { final subscription = _subscriptions.remove(subscriptionKey); if (subscription == null) return; - unsubscribeCount++; - subscribeFilters.remove(subscription.$1); + if (subscription.$1.tags['#h'] != null) { + unsubscribeCount++; + subscribeFilters.remove(subscription.$1); + } else { + visibilitySubscribeFilters.remove(subscription.$1); + } }; } From 5e3c31fd3a870e6c0c390e3fe127434b8f657697 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Mon, 24 Aug 2026 21:00:04 +0100 Subject: [PATCH 2/6] Make DM resurfacing causally ordered and replay-recoverable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Jude review #1: two failures in the DM resurface path. 1. Causal ordering. unhide_dm_recipients cleared every non-null hidden_at with no time comparison, so a recipient who re-hid the DM on another device after a message was accepted could have that newer choice wiped by a delayed resurface. Fence the clear on the triggering message's relay-assigned received_at (AND cm.hidden_at < message_received_at). hidden_at and received_at both come from the server clock (NOW()), so the comparison is against one monotonic authority. 2. Recoverability. Resurface only ran on the fresh-insert path, so a relay that committed the message but died before clearing hidden state never repaired it. Run the (idempotent) fenced resurface on the duplicate/ replay path too, keyed on the message's *original* received_at read back from the DB — so replay self-heals a missed transition without ever clobbering a newer re-hide. Tests: - buzz-db dm::tests (Postgres): older hide clears, newer re-hide survives, sender's own hide is never touched. Wired into the backend-integration CI job by name. - e2e_nostr_interop: replaying the identical message preserves a newer re-hide, proving the duplicate branch fences on the original receive time. Co-authored-by: Mongo <9cfd347903944d5b85aa6c93d2ab67381b978a92a31914bca69998968752a1d7@buzz.block.builderlab.xyz> Signed-off-by: kenny lopez --- .github/workflows/ci.yml | 12 ++ crates/buzz-db/src/dm.rs | 181 ++++++++++++++++++ crates/buzz-db/src/lib.rs | 10 +- crates/buzz-relay/src/handlers/ingest.rs | 78 ++++++-- .../buzz-relay/src/handlers/side_effects.rs | 14 +- .../tests/e2e_nostr_interop.rs | 95 +++++++++ 6 files changed, 368 insertions(+), 22 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 59b9a73ec9b..9e69c9287eb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -744,6 +744,18 @@ jobs: --run-ignored ignored-only env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: DM resurface causal fence + # Verifies unhide_dm_recipients only clears hides causally older than the + # triggering message: an older hide resurfaces, a newer re-hide from + # another device survives a delayed resurface, and the sender's own hide + # is never rewritten. See buzz-db dm::tests. + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-db) and test(/dm::tests::unhide_recipients_/)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Upload relay log if: failure() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 diff --git a/crates/buzz-db/src/dm.rs b/crates/buzz-db/src/dm.rs index e5bfe16fc6d..56ba5180499 100644 --- a/crates/buzz-db/src/dm.rs +++ b/crates/buzz-db/src/dm.rs @@ -453,11 +453,21 @@ pub async fn unhide_dm( /// The sender is deliberately excluded so sending from another surface does /// not rewrite their sidebar preference. Returns only viewers whose hidden /// state changed, allowing the relay to publish targeted visibility snapshots. +/// +/// `message_received_at` is the relay-assigned receive time of the triggering +/// message (`events.received_at`). Only hides that are causally *older* than +/// the message are cleared: a recipient who re-hides the DM on another device +/// after the message was accepted keeps their newer choice. Both `hidden_at` +/// and `received_at` are set from the server clock (`NOW()`), so the comparison +/// is against a single monotonic authority. Because the fence is by receive +/// time (not identity), the update is idempotent — replaying the same message +/// clears the same set and never a newer hide. pub async fn unhide_dm_recipients( pool: &PgPool, community_id: CommunityId, channel_id: Uuid, sender_pubkey: &[u8], + message_received_at: DateTime, ) -> Result>> { let rows = sqlx::query( r#" @@ -469,6 +479,7 @@ pub async fn unhide_dm_recipients( AND cm.pubkey != $3 AND cm.removed_at IS NULL AND cm.hidden_at IS NOT NULL + AND cm.hidden_at < $4 AND c.community_id = cm.community_id AND c.id = cm.channel_id AND c.channel_type = 'dm' @@ -479,6 +490,7 @@ pub async fn unhide_dm_recipients( .bind(community_id.as_uuid()) .bind(channel_id) .bind(sender_pubkey) + .bind(message_received_at) .fetch_all(pool) .await?; @@ -593,4 +605,173 @@ mod tests { let h = compute_participant_hash(&[&a, &b]); assert_eq!(h.len(), 32); } + + // -- Postgres-backed fence tests ------------------------------------------ + // + // `unhide_dm_recipients` must only clear hides that are causally older than + // the triggering message (Jude review #1). These verify the receive-time + // fence directly against real timestamps. + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + + async fn setup_pool() -> PgPool { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + + PgPool::connect(&database_url) + .await + .expect("connect to test DB") + } + + async fn make_test_community(pool: &PgPool) -> CommunityId { + let id = Uuid::new_v4(); + let host = format!("dm-test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(host) + .execute(pool) + .await + .expect("insert test community"); + CommunityId::from_uuid(id) + } + + /// Force a recipient's `hidden_at` to an explicit timestamp so the fence can + /// be exercised deterministically without racing the wall clock. + async fn set_hidden_at( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + pubkey: &[u8], + hidden_at: DateTime, + ) { + sqlx::query( + "UPDATE channel_members SET hidden_at = $4 \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .bind(hidden_at) + .execute(pool) + .await + .expect("set hidden_at"); + } + + async fn current_hidden_at( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + pubkey: &[u8], + ) -> Option> { + sqlx::query( + "SELECT hidden_at FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .fetch_one(pool) + .await + .expect("read hidden_at") + .try_get::>, _>("hidden_at") + .expect("hidden_at column") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn unhide_recipients_clears_hides_older_than_the_message() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let sender = [1u8; 32]; + let recipient = [2u8; 32]; + let dm = create_dm(&pool, community_id, &[&sender, &recipient], &sender) + .await + .expect("create dm"); + + // Recipient hid the DM before the message was received. + let hidden_at = Utc::now(); + set_hidden_at(&pool, community_id, dm.id, &recipient, hidden_at).await; + let message_received_at = hidden_at + chrono::Duration::seconds(1); + + let cleared = + unhide_dm_recipients(&pool, community_id, dm.id, &sender, message_received_at) + .await + .expect("unhide"); + + assert_eq!( + cleared, + vec![recipient.to_vec()], + "recipient must resurface" + ); + assert!( + current_hidden_at(&pool, community_id, dm.id, &recipient) + .await + .is_none(), + "older hide must be cleared" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn unhide_recipients_preserves_a_hide_newer_than_the_message() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let sender = [3u8; 32]; + let recipient = [4u8; 32]; + let dm = create_dm(&pool, community_id, &[&sender, &recipient], &sender) + .await + .expect("create dm"); + + // Recipient re-hid the DM on another device AFTER the message arrived + // (e.g. a delayed replay of that message races the newer user action). + let message_received_at = Utc::now(); + let hidden_at = message_received_at + chrono::Duration::seconds(1); + set_hidden_at(&pool, community_id, dm.id, &recipient, hidden_at).await; + + let cleared = + unhide_dm_recipients(&pool, community_id, dm.id, &sender, message_received_at) + .await + .expect("unhide"); + + assert!( + cleared.is_empty(), + "a hide newer than the message must not be reported as changed" + ); + assert_eq!( + current_hidden_at(&pool, community_id, dm.id, &recipient).await, + Some(hidden_at), + "the newer hide must survive the delayed resurface" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn unhide_recipients_never_touches_the_sender() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let sender = [5u8; 32]; + let recipient = [6u8; 32]; + let dm = create_dm(&pool, community_id, &[&sender, &recipient], &sender) + .await + .expect("create dm"); + + // Both participants have an old hide; only the recipient may be cleared. + let hidden_at = Utc::now(); + set_hidden_at(&pool, community_id, dm.id, &sender, hidden_at).await; + set_hidden_at(&pool, community_id, dm.id, &recipient, hidden_at).await; + let message_received_at = hidden_at + chrono::Duration::seconds(1); + + let cleared = + unhide_dm_recipients(&pool, community_id, dm.id, &sender, message_received_at) + .await + .expect("unhide"); + + assert_eq!(cleared, vec![recipient.to_vec()]); + assert_eq!( + current_hidden_at(&pool, community_id, dm.id, &sender).await, + Some(hidden_at), + "the sender's own hide must never be rewritten from another surface" + ); + } } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index ee092b3c85f..98b802696f5 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2906,8 +2906,16 @@ impl Db { community_id: CommunityId, channel_id: Uuid, sender_pubkey: &[u8], + message_received_at: DateTime, ) -> Result>> { - dm::unhide_dm_recipients(&self.pool, community_id, channel_id, sender_pubkey).await + dm::unhide_dm_recipients( + &self.pool, + community_id, + channel_id, + sender_pubkey, + message_received_at, + ) + .await } /// List the channel IDs of all DMs the given user currently has hidden. diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 50b87ee0f95..795380e1273 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3189,34 +3189,72 @@ async fn ingest_event_inner( } }; - if !was_inserted { - return Ok(IngestResult { - event_id: event_id_hex, - accepted: true, - message: "duplicate:".into(), - }); - } - + // Resurface hidden DMs for the message recipients. This runs on BOTH the + // fresh-insert and duplicate/replay paths: coupling recovery to replay makes + // the visibility transition self-healing if the relay committed the message + // but died before clearing hidden state (Jude review #1, recoverability). + // + // The fence is the message's *original* received_at. On a fresh insert that + // is `stored_event.received_at`; on a replay `stored_event` carries the new + // replay time, so we read the persisted receive time back from the DB. This + // keeps a recipient who re-hid the DM after the message was first accepted + // from having that newer choice cleared by a later replay (Jude review #1, + // causal ordering). The clear itself is idempotent, so re-running is safe. if is_human_visible_message_kind(kind_u32) { if let (Some(ch_id), Some(channel)) = (channel_id, channel_row.as_ref()) { if channel.channel_type == "dm" { - let sender = effective_message_author(&event, &state.relay_keypair.public_key()); - if let Err(error) = - crate::handlers::side_effects::resurface_dm_for_message_recipients( - tenant, state, ch_id, &sender, - ) - .await - { - error!( - event_id = %event_id_hex, - channel_id = %ch_id, - "Failed to resurface DM for message recipients: {error}" - ); + let message_received_at = if was_inserted { + Some(stored_event.received_at) + } else { + match state + .db + .get_event_by_id(tenant.community(), event.id.as_bytes()) + .await + { + Ok(Some(original)) => Some(original.received_at), + Ok(None) => None, + Err(error) => { + error!( + event_id = %event_id_hex, + channel_id = %ch_id, + "Failed to load original message for DM resurface replay: {error}" + ); + None + } + } + }; + if let Some(message_received_at) = message_received_at { + let sender = + effective_message_author(&event, &state.relay_keypair.public_key()); + if let Err(error) = + crate::handlers::side_effects::resurface_dm_for_message_recipients( + tenant, + state, + ch_id, + &sender, + message_received_at, + ) + .await + { + error!( + event_id = %event_id_hex, + channel_id = %ch_id, + "Failed to resurface DM for message recipients: {error}" + ); + } } } } } + if !was_inserted { + return Ok(IngestResult { + event_id: event_id_hex, + accepted: true, + message: "duplicate:".into(), + }); + } + if crate::handlers::side_effects::is_side_effect_kind(kind_u32) { if let Err(e) = crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state) diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index b51dd79016a..69980b96c73 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3473,15 +3473,27 @@ pub async fn publish_dm_visibility_snapshot( /// /// Hidden state is per viewer, so only active members other than the effective /// author are changed. A fresh snapshot makes the update visible across clients. +/// +/// `message_received_at` fences the clear so only hides older than the message +/// are lifted (see [`buzz_db::Database::unhide_dm_recipients`]). Passing the +/// message's original `received_at` also makes this safe to re-run on the +/// duplicate/replay path: the operation is idempotent, so a crash between the +/// accepted message and the resurface self-heals when the message is replayed. pub async fn resurface_dm_for_message_recipients( tenant: &TenantContext, state: &Arc, channel_id: Uuid, sender_pubkey: &[u8], + message_received_at: chrono::DateTime, ) -> anyhow::Result<()> { let recipients = state .db - .unhide_dm_recipients(tenant.community(), channel_id, sender_pubkey) + .unhide_dm_recipients( + tenant.community(), + channel_id, + sender_pubkey, + message_received_at, + ) .await?; let mut failures = Vec::new(); diff --git a/crates/buzz-test-client/tests/e2e_nostr_interop.rs b/crates/buzz-test-client/tests/e2e_nostr_interop.rs index 8c1c49b6b2f..1766828be24 100644 --- a/crates/buzz-test-client/tests/e2e_nostr_interop.rs +++ b/crates/buzz-test-client/tests/e2e_nostr_interop.rs @@ -191,6 +191,36 @@ async fn post_signed_event(keys: &Keys, kind: u16, tags: Vec) { ); } +/// Submit an already-signed message event via REST and assert acceptance. +/// +/// Unlike [`send_rest_message_with_kind`], the caller owns the signed event so +/// it can be submitted verbatim more than once — the second submission lands on +/// the relay's duplicate/replay branch, which is what the recovery test needs. +/// The relay reports `accepted: true` for both the fresh insert and the +/// duplicate (`message: "duplicate:"`), so only `accepted` is asserted. +async fn post_signed_message_event(keys: &Keys, event: &nostr::Event) { + let client = reqwest::Client::new(); + let pubkey_hex = keys.public_key().to_hex(); + let resp = client + .post(format!("{}/events", relay_http_url())) + .header("X-Pubkey", &pubkey_hex) + .header("Content-Type", "application/json") + .body(serde_json::to_string(event).unwrap()) + .send() + .await + .expect("submit signed message event"); + assert!( + resp.status().is_success(), + "signed message submit failed: {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.expect("parse event response"); + assert!( + body["accepted"].as_bool().unwrap_or(false), + "signed message not accepted: {body}" + ); +} + /// Query the relay for the thread replies recorded under `root_event_id`. /// /// Uses `POST /query` with the `depth_limit` extension field, which the relay's @@ -1410,6 +1440,71 @@ async fn test_nipdv_group_message_resurfaces_all_hidden_recipients() { } } +/// Replaying the exact same accepted DM message must NOT clobber a re-hide the +/// recipient performed after the message first arrived. This is the recovery +/// path from Jude review #1: resurface runs on the duplicate branch too (so a +/// relay that committed the message but died before clearing hidden state +/// self-heals on replay), but it must fence on the message's *original* +/// received_at — never the replay time — so a newer user action survives. +#[tokio::test] +#[ignore] +async fn test_nipdv_message_replay_preserves_a_newer_rehide() { + let url = relay_url(); + let keys_a = Keys::generate(); + let keys_b = Keys::generate(); + let b_pubkey_hex = keys_b.public_key().to_hex(); + let channel_id = create_dm(&keys_a, &b_pubkey_hex).await; + + // B hides the DM, then A sends a message. The message resurfaces it for B. + post_signed_event( + &keys_b, + 41012, + vec![Tag::parse(["h", &channel_id]).unwrap()], + ) + .await; + + // Sign the message once so it can be submitted verbatim twice — the second + // POST exercises the relay's duplicate/replay branch. + let message = EventBuilder::new(Kind::Custom(9), "inbound activity") + .tags(vec![Tag::parse(["h", &channel_id]).unwrap()]) + .sign_with_keys(&keys_a) + .unwrap(); + post_signed_message_event(&keys_a, &message).await; + + let mut client_b = BuzzTestClient::connect(&url, &keys_b) + .await + .expect("client B connect"); + let after_message = read_hidden_dms(&mut client_b, &b_pubkey_hex).await; + assert!( + !after_message.contains(&channel_id), + "the message must resurface the DM for B; B sees: {after_message:?}" + ); + client_b.disconnect().await.expect("disconnect B"); + + // B re-hides the DM AFTER the message was accepted. This is the newer user + // action that a delayed replay must not undo. + post_signed_event( + &keys_b, + 41012, + vec![Tag::parse(["h", &channel_id]).unwrap()], + ) + .await; + + // Replay the identical message. The relay dedupes it (duplicate branch), and + // the fenced resurface must leave B's newer hide intact. + post_signed_message_event(&keys_a, &message).await; + + let mut client_b = BuzzTestClient::connect(&url, &keys_b) + .await + .expect("client B reconnect"); + let after_replay = read_hidden_dms(&mut client_b, &b_pubkey_hex).await; + assert!( + after_replay.contains(&channel_id), + "replaying the original message must not clear B's newer re-hide; B sees: {after_replay:?}" + ); + client_b.disconnect().await.expect("disconnect B"); +} + /// NIP-DV monotonicity regression: a hide immediately followed by a re-open /// within the same wall-clock second must still leave the re-open authoritative. /// From 42ff8cced079e4c977dd4b77d7a918f6a8cdb9f0 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Tue, 25 Aug 2026 07:33:29 +0100 Subject: [PATCH 3/6] Fix flaky DM resurface fence tests The three Postgres fence tests in buzz-db asserted hidden_at equality against raw Utc::now() values, but timestamptz stores microsecond precision, so the nanosecond tail was truncated on round-trip and the equality assertions failed in CI. Truncate to microseconds up front. The e2e replay test (test_nipdv_message_replay_preserves_a_newer_rehide) posted two identical kind:41012 hides from B. A nostr event id is sha256 over second-resolution created_at, so when both hides land in the same wall-clock second they share an id and the relay dedupes the second one before re-running hide_dm -- B's re-hide silently vanished and the assertion saw an empty hidden set. Backdate the first hide to force distinct ids, matching the guard create_dm already uses. Both are test-authoring defects; the fence and resurface logic are unchanged. Co-authored-by: Mongo <9cfd347903944d5b85aa6c93d2ab67381b978a92a31914bca69998968752a1d7@buzz.block.builderlab.xyz> Signed-off-by: kenny lopez --- crates/buzz-db/src/dm.rs | 15 +++++++++--- .../tests/e2e_nostr_interop.rs | 23 ++++++++++++++++++- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/crates/buzz-db/src/dm.rs b/crates/buzz-db/src/dm.rs index 56ba5180499..e3e9e623a72 100644 --- a/crates/buzz-db/src/dm.rs +++ b/crates/buzz-db/src/dm.rs @@ -614,6 +614,15 @@ mod tests { const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + /// Postgres `timestamptz` stores microsecond precision, so a raw + /// `Utc::now()` (nanoseconds) will not round-trip equal. Truncate to + /// microseconds up front so equality assertions compare like for like + /// while the seconds-apart fence arithmetic stays intact. + fn now_micros() -> DateTime { + use chrono::SubsecRound; + Utc::now().trunc_subsecs(6) + } + async fn setup_pool() -> PgPool { let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) @@ -690,7 +699,7 @@ mod tests { .expect("create dm"); // Recipient hid the DM before the message was received. - let hidden_at = Utc::now(); + let hidden_at = now_micros(); set_hidden_at(&pool, community_id, dm.id, &recipient, hidden_at).await; let message_received_at = hidden_at + chrono::Duration::seconds(1); @@ -725,7 +734,7 @@ mod tests { // Recipient re-hid the DM on another device AFTER the message arrived // (e.g. a delayed replay of that message races the newer user action). - let message_received_at = Utc::now(); + let message_received_at = now_micros(); let hidden_at = message_received_at + chrono::Duration::seconds(1); set_hidden_at(&pool, community_id, dm.id, &recipient, hidden_at).await; @@ -757,7 +766,7 @@ mod tests { .expect("create dm"); // Both participants have an old hide; only the recipient may be cleared. - let hidden_at = Utc::now(); + let hidden_at = now_micros(); set_hidden_at(&pool, community_id, dm.id, &sender, hidden_at).await; set_hidden_at(&pool, community_id, dm.id, &recipient, hidden_at).await; let message_received_at = hidden_at + chrono::Duration::seconds(1); diff --git a/crates/buzz-test-client/tests/e2e_nostr_interop.rs b/crates/buzz-test-client/tests/e2e_nostr_interop.rs index 1766828be24..8223b9a06a0 100644 --- a/crates/buzz-test-client/tests/e2e_nostr_interop.rs +++ b/crates/buzz-test-client/tests/e2e_nostr_interop.rs @@ -165,10 +165,27 @@ async fn create_dm(requester_keys: &Keys, other_pubkey_hex: &str) -> String { /// Submit a signed command event via REST and assert it was accepted. async fn post_signed_event(keys: &Keys, kind: u16, tags: Vec) { + post_signed_event_created_at(keys, kind, tags, nostr::Timestamp::now()).await; +} + +/// Like [`post_signed_event`] but with an explicit `created_at`. A nostr event +/// id is `sha256(pubkey, created_at, kind, tags, content)` at second +/// resolution, so two otherwise-identical command events posted in the same +/// wall-clock second collide on id and the relay dedupes the second one +/// (`command_executor` returns `PersistResult::Duplicate` before re-running the +/// mutation). Callers that must post the same command twice as distinct actions +/// backdate the first to force distinct ids — the same guard `create_dm` uses. +async fn post_signed_event_created_at( + keys: &Keys, + kind: u16, + tags: Vec, + created_at: nostr::Timestamp, +) { let client = reqwest::Client::new(); let pubkey_hex = keys.public_key().to_hex(); let event = EventBuilder::new(Kind::Custom(kind), "") .tags(tags) + .custom_created_at(created_at) .sign_with_keys(keys) .unwrap(); let resp = client @@ -1456,10 +1473,14 @@ async fn test_nipdv_message_replay_preserves_a_newer_rehide() { let channel_id = create_dm(&keys_a, &b_pubkey_hex).await; // B hides the DM, then A sends a message. The message resurfaces it for B. - post_signed_event( + // Backdate this first hide so it cannot collide on event id with the + // identical re-hide posted later in the same wall-clock second (which the + // relay would dedupe, silently dropping the second hide). + post_signed_event_created_at( &keys_b, 41012, vec![Tag::parse(["h", &channel_id]).unwrap()], + nostr::Timestamp::from(nostr::Timestamp::now().as_secs() - 10), ) .await; From 9f8403847cefb546e96ac744b63f0e859aee0fcb Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Tue, 25 Aug 2026 08:57:14 +0100 Subject: [PATCH 4/6] Resurface hidden DMs from workflow sends and make publication retryable Jude review #2: workflow SendMessage persists a real human-visible kind:9 without entering generic ingest, so a hidden recipient stayed hidden when the DM came from a workflow/agent. Invoke the same causally-fenced, replay-recoverable resurface from the workflow writer, gated on DM channel_type and keyed on the message's original received_at (read back on the duplicate/replay branch) so a newer re-hide is never cleared. Jude review #1 (retryable-publication slice): resurface published snapshots only for the rows unhide_dm_recipients changed. A replay clears zero rows (hidden_at already NULL), so a snapshot whose original publish failed was never retried and the client stayed hidden indefinitely. Resurface now republishes for every active non-sender recipient; publish_dm_visibility_snapshot gains a per-viewer drift guard that skips the replace when the last published snapshot already conveys the canonical hidden set. Together this repairs a lagging snapshot on replay while never churning converged viewers or minting an empty snapshot for a never-hidden member. The durable cross-process ordering/generation invariant Jude also requested (TOCTOU between a paused stale-hide publisher and a resurface) is a larger design change and is intentionally out of scope for this slice. Tests (Postgres-gated, matching the existing DM suite): - workflow_dm_message_resurfaces_hidden_recipient: hidden recipient + workflow kind:9 DM send -> recipient resurfaced, sender's hide preserved, recipient snapshot no longer lists the DM. - resurface_repairs_a_stale_snapshot_on_replay: canonical clear committed but snapshot publish 'failed' -> replay republishes and repairs the lag. Co-authored-by: Mongo <9cfd347903944d5b85aa6c93d2ab67381b978a92a31914bca69998968752a1d7@buzz.block.builderlab.xyz> Signed-off-by: kenny lopez --- .../buzz-relay/src/handlers/side_effects.rs | 85 ++++-- crates/buzz-relay/src/workflow_sink.rs | 278 +++++++++++++++++- 2 files changed, 343 insertions(+), 20 deletions(-) diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 69980b96c73..1cff9bdd4ff 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3395,6 +3395,44 @@ pub async fn publish_dm_visibility_snapshot( let hidden = state.db.list_hidden_dms(tenant.community(), viewer).await?; let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); + // Read the latest snapshot once: it feeds both the drift check below and the + // monotonic `created_at` bump. A missing snapshot conveys the empty set. + let existing = state + .db + .query_events(&buzz_db::event::EventQuery { + kinds: Some(vec![KIND_DM_VISIBILITY as i32]), + pubkey: Some(state.relay_keypair.public_key().to_bytes().to_vec()), + d_tag: Some(viewer_hex.clone()), + limit: Some(1), + ..buzz_db::event::EventQuery::for_community(tenant.community()) + }) + .await + .unwrap_or_default(); + + // Drift guard: skip the replace when the last published snapshot already + // conveys the canonical hidden set. This makes republishing idempotent, so + // callers can safely re-run for every active recipient (Jude review #1, + // retryable publication) without churning unchanged viewers or minting an + // empty snapshot for a viewer who has never hidden a DM. When canonical and + // conveyed diverge — e.g. a replay repairing a resurface whose original + // publish failed — the snapshot is republished. + let canonical: std::collections::HashSet = + hidden.iter().map(|c| c.to_string()).collect(); + let conveyed: std::collections::HashSet = existing + .first() + .map(|e| { + e.event + .tags + .iter() + .filter(|t| t.kind().to_string() == "h") + .filter_map(|t| t.content().map(str::to_owned)) + .collect() + }) + .unwrap_or_default(); + if canonical == conveyed { + return Ok(()); + } + let mut tags: Vec = Vec::with_capacity(hidden.len() + 2); tags.push( Tag::parse(["d", &viewer_hex]) @@ -3421,23 +3459,10 @@ pub async fn publish_dm_visibility_snapshot( .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs(); - let ts = { - let existing = state - .db - .query_events(&buzz_db::event::EventQuery { - kinds: Some(vec![KIND_DM_VISIBILITY as i32]), - pubkey: Some(state.relay_keypair.public_key().to_bytes().to_vec()), - d_tag: Some(viewer_hex.clone()), - limit: Some(1), - ..buzz_db::event::EventQuery::for_community(tenant.community()) - }) - .await - .unwrap_or_default(); - existing - .first() - .map(|e| (e.event.created_at.as_secs() + 1).max(now)) - .unwrap_or(now) - }; + let ts = existing + .first() + .map(|e| (e.event.created_at.as_secs() + 1).max(now)) + .unwrap_or(now); let event = EventBuilder::new(Kind::Custom(KIND_DM_VISIBILITY as u16), "") .tags(tags) @@ -3472,13 +3497,22 @@ pub async fn publish_dm_visibility_snapshot( /// Resurface a DM for recipients of a newly accepted message. /// /// Hidden state is per viewer, so only active members other than the effective -/// author are changed. A fresh snapshot makes the update visible across clients. +/// author are cleared. A fresh snapshot makes the update visible across clients. /// /// `message_received_at` fences the clear so only hides older than the message /// are lifted (see [`buzz_db::Database::unhide_dm_recipients`]). Passing the /// message's original `received_at` also makes this safe to re-run on the /// duplicate/replay path: the operation is idempotent, so a crash between the /// accepted message and the resurface self-heals when the message is replayed. +/// +/// Snapshot publication targets *every* active non-sender recipient, not only +/// the rows the clear changed. `unhide_dm_recipients` returns zero rows on a +/// replay (the hide was already cleared), so publishing only for changed rows +/// would never retry a snapshot whose original publish failed — leaving the +/// client hidden indefinitely (Jude review #1, retryable publication). The +/// per-viewer drift guard in [`publish_dm_visibility_snapshot`] keeps this +/// idempotent: viewers whose latest snapshot already matches canonical state +/// are skipped, so the wider fan-out never churns unchanged recipients. pub async fn resurface_dm_for_message_recipients( tenant: &TenantContext, state: &Arc, @@ -3486,7 +3520,7 @@ pub async fn resurface_dm_for_message_recipients( sender_pubkey: &[u8], message_received_at: chrono::DateTime, ) -> anyhow::Result<()> { - let recipients = state + state .db .unhide_dm_recipients( tenant.community(), @@ -3496,6 +3530,19 @@ pub async fn resurface_dm_for_message_recipients( ) .await?; + // Republish for every active recipient other than the sender. Combined with + // the drift guard this is a no-op for viewers already converged, and repairs + // any viewer whose snapshot lags the canonical hidden set (e.g. a resurface + // whose publish previously failed and is now reached on replay). + let recipients: Vec> = state + .db + .get_members(tenant.community(), channel_id) + .await? + .into_iter() + .map(|m| m.pubkey) + .filter(|pubkey| pubkey.as_slice() != sender_pubkey) + .collect(); + let mut failures = Vec::new(); for recipient in recipients { if let Err(error) = publish_dm_visibility_snapshot(tenant, state, &recipient).await { diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 8ce23a2e8ea..5278c7009b0 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -13,7 +13,7 @@ use buzz_core::tenant::CommunityId; use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; use chrono::Utc; use nostr::{EventBuilder, Kind, Tag}; -use tracing::info; +use tracing::{error, info}; use uuid::Uuid; use crate::handlers::event::dispatch_persistent_event; @@ -426,6 +426,57 @@ impl ActionSink for RelayActionSink { } } + // Resurface hidden DMs for the recipients of this message. Workflow + // sends persist a real human-visible kind:9 without entering generic + // ingest, so without this call a hidden recipient stays hidden when + // the DM comes from a workflow/agent (Jude review #2). Runs on both + // the fresh-insert and replay paths and is fenced by the message's + // *original* received_at, matching the ingest path: on a fresh insert + // that is `stored_event.received_at`; on a replay `stored_event` + // carries the new replay time, so the persisted receive time is read + // back so a recipient's newer re-hide is never cleared. Best-effort: + // a failure here must not fail the workflow send. + if channel.channel_type == "dm" { + let message_received_at = if was_inserted { + Some(stored_event.received_at) + } else { + match state + .db + .get_event_by_id(tenant.community(), event.id.as_bytes()) + .await + { + Ok(Some(original)) => Some(original.received_at), + Ok(None) => None, + Err(error) => { + error!( + event_id = %event_id_hex, + channel_id = %channel_id_canonical, + "Workflow SendMessage: failed to load original message for DM resurface replay: {error}" + ); + None + } + } + }; + if let Some(message_received_at) = message_received_at { + if let Err(error) = + crate::handlers::side_effects::resurface_dm_for_message_recipients( + &tenant, + &state, + channel_uuid, + &author_pubkey_bytes, + message_received_at, + ) + .await + { + error!( + event_id = %event_id_hex, + channel_id = %channel_id_canonical, + "Workflow SendMessage: failed to resurface DM for message recipients: {error}" + ); + } + } + } + Ok(event_id_hex) }) } @@ -676,6 +727,51 @@ mod integration_tests { Arc::new(state) } + /// Build the `TenantContext` for a community the same way `send_message` + /// does (host is label-only; the community is authoritative). + async fn tenant_for( + state: &Arc, + community: CommunityId, + ) -> buzz_core::tenant::TenantContext { + let host = state + .db + .lookup_community_host(community) + .await + .expect("lookup host") + .expect("community mapped to host"); + buzz_core::tenant::TenantContext::resolved(community, host) + } + + /// The `h` tags of a viewer's latest relay-signed NIP-DV snapshot. + async fn read_snapshot_h_tags( + state: &Arc, + community: CommunityId, + viewer_hex: &str, + ) -> Vec { + let events = state + .db + .query_events(&buzz_db::event::EventQuery { + kinds: Some(vec![buzz_core::kind::KIND_DM_VISIBILITY as i32]), + pubkey: Some(state.relay_keypair.public_key().to_bytes().to_vec()), + d_tag: Some(viewer_hex.to_owned()), + limit: Some(1), + ..buzz_db::event::EventQuery::for_community(community) + }) + .await + .expect("query snapshot"); + events + .first() + .map(|e| { + e.event + .tags + .iter() + .filter(|t| t.kind().to_string() == "h") + .filter_map(|t| t.content().map(str::to_owned)) + .collect() + }) + .unwrap_or_default() + } + #[tokio::test] #[ignore = "requires Postgres"] async fn workflow_send_message_p_tags_mentioned_member() { @@ -888,6 +984,186 @@ mod integration_tests { assert_eq!(meta.root_event_id.as_deref(), Some(root_bytes.as_slice())); } + /// Jude review #2: a workflow-authored DM message must resurface the hidden + /// DM for its recipients while preserving the sender's own hide. The workflow + /// path persists a real kind:9 without entering generic ingest, so the + /// resurface must be invoked from the workflow writer. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_dm_message_resurfaces_hidden_recipient() { + let state = test_state().await; + + let author = nostr::Keys::generate(); + let author_hex = author.public_key().to_hex(); + let author_bytes = author.public_key().to_bytes().to_vec(); + let recipient = nostr::Keys::generate(); + let recipient_bytes = recipient.public_key().to_bytes().to_vec(); + + let host = format!("wf-dm-resurface-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &author_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + + // A DM between the workflow owner (author) and the recipient. Both are + // active members; the channel is private. + let dm = state + .db + .create_dm( + community, + &[author_bytes.as_slice(), recipient_bytes.as_slice()], + &author_bytes, + ) + .await + .expect("create dm"); + + // Both participants hide the DM. + state + .db + .hide_dm(community, dm.id, &author_bytes) + .await + .expect("hide for author"); + state + .db + .hide_dm(community, dm.id, &recipient_bytes) + .await + .expect("hide for recipient"); + + // The workflow sends a DM message authored by the owner. + let sink = RelayActionSink::new(&state); + sink.send_message( + community, + &dm.id.to_string(), + "workflow ping", + &author_hex, + None, + ) + .await + .expect("workflow send_message into DM"); + + // The recipient's hide is cleared; the sender's own hide survives. + let recipient_hidden = state + .db + .list_hidden_dms(community, &recipient_bytes) + .await + .expect("list recipient hidden"); + assert!( + !recipient_hidden.contains(&dm.id), + "workflow DM must resurface the DM for the recipient; recipient sees: {recipient_hidden:?}" + ); + + let author_hidden = state + .db + .list_hidden_dms(community, &author_bytes) + .await + .expect("list author hidden"); + assert!( + author_hidden.contains(&dm.id), + "workflow DM must preserve the sender's own hide; author sees: {author_hidden:?}" + ); + + // The relay published a NIP-DV snapshot for the resurfaced recipient + // reflecting the now-empty hidden set. + let hidden_h_tags = + read_snapshot_h_tags(&state, community, &hex::encode(&recipient_bytes)).await; + assert!( + !hidden_h_tags.contains(&dm.id.to_string()), + "recipient snapshot must no longer list the resurfaced DM; h tags: {hidden_h_tags:?}" + ); + } + + /// Jude review #1 (retryable publication): a resurface whose canonical clear + /// committed but whose snapshot publish failed must self-heal on replay. The + /// clear reports zero changed rows the second time (`hidden_at` is already + /// NULL), so publishing only for changed rows would never republish — the + /// client stays hidden forever. Resurface now republishes for every active + /// recipient, and the drift guard repairs the lagging snapshot. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn resurface_repairs_a_stale_snapshot_on_replay() { + let state = test_state().await; + + let author = nostr::Keys::generate(); + let author_hex = author.public_key().to_hex(); + let author_bytes = author.public_key().to_bytes().to_vec(); + let recipient = nostr::Keys::generate(); + let recipient_bytes = recipient.public_key().to_bytes().to_vec(); + let recipient_hex = hex::encode(&recipient_bytes); + + let host = format!("wf-dm-repair-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &author_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + + let dm = state + .db + .create_dm( + community, + &[author_bytes.as_slice(), recipient_bytes.as_slice()], + &author_bytes, + ) + .await + .expect("create dm"); + + // The recipient hid the DM and their snapshot reflects it. + state + .db + .hide_dm(community, dm.id, &recipient_bytes) + .await + .expect("hide for recipient"); + crate::handlers::side_effects::publish_dm_visibility_snapshot( + &tenant_for(&state, community).await, + &state, + &recipient_bytes, + ) + .await + .expect("publish stale snapshot"); + + // Simulate a resurface whose canonical clear committed but whose snapshot + // publish failed: clear the hide directly, WITHOUT republishing. The + // recipient's latest snapshot now lags canonical state (still lists the + // DM as hidden) — the exact stuck state Jude flagged. + state + .db + .unhide_dm(community, dm.id, &recipient_bytes) + .await + .expect("clear recipient hide"); + let stale = read_snapshot_h_tags(&state, community, &recipient_hex).await; + assert!( + stale.contains(&dm.id.to_string()), + "precondition: snapshot must still lag canonical state; got: {stale:?}" + ); + + // Replay resurface. The clear now changes zero rows, but publication must + // still run for the recipient and repair the lagging snapshot. + crate::handlers::side_effects::resurface_dm_for_message_recipients( + &tenant_for(&state, community).await, + &state, + dm.id, + &author_bytes, + chrono::Utc::now(), + ) + .await + .expect("replay resurface"); + + let repaired = read_snapshot_h_tags(&state, community, &recipient_hex).await; + assert!( + !repaired.contains(&dm.id.to_string()), + "replay must republish the recipient snapshot to drop the resurfaced DM; got: {repaired:?}" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn workflow_replies_recover_metadata_less_parent_ancestry() { From d28c8b0f4e9f8a8c54352d065ed9e632544bc7d8 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Tue, 25 Aug 2026 10:12:35 +0100 Subject: [PATCH 5/6] Propagate DM visibility snapshot-read errors instead of masking as convergence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The drift guard read the latest published snapshot with `.await.unwrap_or_default()`, turning a transient DB read *error* into an empty conveyed set. When canonical state was also empty (a resurface just cleared the hide), the guard mistook the fabricated empty for convergence and returned Ok(()) without republishing — leaving a stale snapshot that keeps the DM hidden on the client across reconnect/restart. Propagate the error with `?` so the per-recipient caller in resurface_dm_for_message_recipients records it and the message replay path retries. This matches the sibling `list_hidden_dms` read two lines above, which already uses `?`. Co-authored-by: Mongo <9cfd347903944d5b85aa6c93d2ab67381b978a92a31914bca69998968752a1d7@buzz.block.builderlab.xyz> Signed-off-by: kenny lopez --- crates/buzz-relay/src/handlers/side_effects.rs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 1cff9bdd4ff..90de1091f6e 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3396,7 +3396,12 @@ pub async fn publish_dm_visibility_snapshot( let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); // Read the latest snapshot once: it feeds both the drift check below and the - // monotonic `created_at` bump. A missing snapshot conveys the empty set. + // monotonic `created_at` bump. A missing snapshot (empty `Ok`) conveys the + // empty set, but a read *error* must propagate: swallowing it as `[]` would + // let the drift guard below mistake a transient failure for convergence and + // report `Ok(())` while a stale snapshot survives, stranding a hidden DM on + // the client forever (Jude review: snapshot-read failure misreported as + // convergence). Propagating instead makes the caller log and retry. let existing = state .db .query_events(&buzz_db::event::EventQuery { @@ -3406,8 +3411,7 @@ pub async fn publish_dm_visibility_snapshot( limit: Some(1), ..buzz_db::event::EventQuery::for_community(tenant.community()) }) - .await - .unwrap_or_default(); + .await?; // Drift guard: skip the replace when the last published snapshot already // conveys the canonical hidden set. This makes republishing idempotent, so From 758a0507871d595f28c901aab4f147fbe804c408 Mon Sep 17 00:00:00 2001 From: kenny lopez Date: Tue, 25 Aug 2026 20:13:54 +0100 Subject: [PATCH 6/6] fix(dm-visibility): harden NIP-DV hidden-DM resurface path against reviewer-flagged defects Address Wess/Jude review defects on the hidden-DM visibility/resurface path. Each is an in-scope correctness fix to the mechanism this PR builds; none change product intent or UI. The #6769 durable cross-process convergence invariant remains an intentional out-of-scope follow-up (bug #6316 stays open, tracked by #6769). A. Clock-skew fence (buzz-db/event.rs): received_at was assigned by the relay process clock (Utc::now()) while hide_dm's hidden_at uses the DB clock (NOW()), so the causal resurface fence compared two clocks. Insert now writes NOW() and RETURNs the value, sourcing both sides from the one database clock. Adds a PG-gated regression bracketing received_at between two DB NOW() reads. B. Mobile replay loop (dm_visibility.dart): the limit:1 snapshot replay could fire before subscribe() resolved, kicking refresh() -> version bump -> resubscribe -> identical replay, unbounded for any user with a stored snapshot. Dedup on snapshot event id (reset on relay/identity change and clear). Adds a regression. C. Huddle invites never resurfaced (buzz-core/kind.rs): KIND_HUDDLE_STARTED is rendered as visible content and treated as a notifiable DM invite but was absent from is_human_visible_message_kind, so it never resurfaced a hidden DM. Added to the trigger set; other huddle lifecycle kinds stay excluded. Test updated. D. Fail-open snapshot read (desktop fetch.rs, mobile channel_directory.dart): a transient kind:30622 read failure was treated as 'nothing hidden', re-exposing every hidden DM until the next good refresh. Now fails closed by propagating the error so the caller preserves the last rendered list. Adds a mobile regression. E. Desktop in-flight invalidation race (useMembershipNotifications.ts): direct channelsQueryKey invalidation could be silently undone by an in-flight get_channels, dropping the resurface signal. Routed through the idle-aware trailing debounce (refreshChannelsWhenIdle) as useLiveChannelUpdates does, with cleanup-time cancel(). Test updated plus an in-flight re-arm regression. Verified locally at this tree: cargo fmt --check, clippy (buzz-core, buzz-db), file-size-check; buzz-core kind tests, buzz-db event tests (incl. PG-gated), buzz-relay resurface integration tests; desktop node --test useMembershipNotifications; mobile flutter analyze + full channels_provider_test suite. All green. Co-authored-by: Mongo <9cfd347903944d5b85aa6c93d2ab67381b978a92a31914bca69998968752a1d7@buzz.block.builderlab.xyz> Signed-off-by: kenny lopez --- crates/buzz-core/src/kind.rs | 23 +++- crates/buzz-db/src/event.rs | 90 +++++++++++-- .../src-tauri/src/commands/channels/fetch.rs | 43 ++++--- .../useMembershipNotifications.test.mjs | 99 ++++++++++++++- .../channels/useMembershipNotifications.ts | 31 ++++- .../features/channels/channel_directory.dart | 31 +++-- .../channels_provider/dm_visibility.dart | 19 +++ .../channels/channels_provider_test.dart | 120 ++++++++++++++++++ 8 files changed, 409 insertions(+), 47 deletions(-) diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index 89651507636..a775e28f790 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -794,10 +794,20 @@ pub const fn is_workflow_execution_kind(kind: u32) -> bool { /// /// Reactions, edits, deletions, and system events are deliberately excluded: /// they must not resurface a direct message without a new human-visible message. +/// +/// `KIND_HUDDLE_STARTED` is included: clients render the huddle-start card as +/// visible timeline content and desktop treats it as a notifiable DM invitation, +/// so a hidden DM must resurface to deliver the (time-sensitive) invite. The +/// other huddle lifecycle kinds (joined/left/ended/reaction) are not message +/// content and stay excluded. pub const fn is_human_visible_message_kind(kind: u32) -> bool { matches!( kind, - KIND_STREAM_MESSAGE | KIND_STREAM_MESSAGE_V2 | KIND_FORUM_POST | KIND_FORUM_COMMENT + KIND_STREAM_MESSAGE + | KIND_STREAM_MESSAGE_V2 + | KIND_FORUM_POST + | KIND_FORUM_COMMENT + | KIND_HUDDLE_STARTED ) } @@ -951,11 +961,20 @@ mod tests { KIND_STREAM_MESSAGE_V2, KIND_FORUM_POST, KIND_FORUM_COMMENT, + KIND_HUDDLE_STARTED, ] { assert!(is_human_visible_message_kind(kind), "kind {kind}"); } - for kind in [KIND_REACTION, KIND_STREAM_MESSAGE_EDIT, KIND_DELETION] { + for kind in [ + KIND_REACTION, + KIND_STREAM_MESSAGE_EDIT, + KIND_DELETION, + KIND_HUDDLE_PARTICIPANT_JOINED, + KIND_HUDDLE_PARTICIPANT_LEFT, + KIND_HUDDLE_ENDED, + KIND_HUDDLE_REACTION, + ] { assert!(!is_human_visible_message_kind(kind), "kind {kind}"); } } diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index dcfc97c8447..01ea4c05a29 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -319,14 +319,19 @@ async fn insert_event_on( let created_at_secs = event.created_at.as_secs() as i64; let created_at = DateTime::from_timestamp(created_at_secs, 0) .ok_or(DbError::InvalidTimestamp(created_at_secs))?; - let received_at = Utc::now(); let d_tag = extract_d_tag(event); let not_before = extract_not_before(event); - let result = sqlx::query( + // `received_at` is assigned by the database clock (`NOW()`) and read back, + // not by the relay process clock. The DM resurface fence compares this + // value against `channel_members.hidden_at`, which is also written with + // `NOW()` (see `hide_dm`); sourcing both from the single database clock is + // what makes that causal comparison sound under relay/DB clock skew. + let inserted_received_at: Option> = sqlx::query_scalar( r#" INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), $9, $10, $11) ON CONFLICT DO NOTHING + RETURNING received_at "#, ) .bind(community_id.as_uuid()) @@ -337,14 +342,17 @@ async fn insert_event_on( .bind(&tags_json) .bind(&event.content) .bind(sig_bytes.as_slice()) - .bind(received_at) .bind(channel_id) .bind(d_tag.as_deref()) .bind(not_before) - .execute(connection) + .fetch_optional(connection) .await?; - let was_inserted = result.rows_affected() > 0; + let was_inserted = inserted_received_at.is_some(); + // On a conflict (duplicate/replay) no row is returned; the value is only + // informational here because callers gate on `was_inserted`, and the + // resurface replay path re-reads the original event's `received_at`. + let received_at = inserted_received_at.unwrap_or_else(Utc::now); Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), @@ -1178,15 +1186,19 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( let created_at_secs = event.created_at.as_secs() as i64; let created_at = DateTime::from_timestamp(created_at_secs, 0) .ok_or(DbError::InvalidTimestamp(created_at_secs))?; - let received_at = Utc::now(); let d_tag = extract_d_tag(event); let not_before = extract_not_before(event); - let result = sqlx::query( + // `received_at` is assigned by the database clock (`NOW()`) and read back, + // not by the relay process clock, so the DM resurface fence compares it + // against `channel_members.hidden_at` (also `NOW()`) under one clock. See + // `insert_event_on` for the full rationale. + let inserted_received_at: Option> = sqlx::query_scalar( r#" INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW(), $9, $10, $11) ON CONFLICT DO NOTHING + RETURNING received_at "#, ) .bind(community_id.as_uuid()) @@ -1197,14 +1209,17 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( .bind(&tags_json) .bind(&event.content) .bind(sig_bytes.as_slice()) - .bind(received_at) .bind(channel_id) .bind(d_tag.as_deref()) .bind(not_before) - .execute(&mut **tx) + .fetch_optional(&mut **tx) .await?; - let was_inserted = result.rows_affected() > 0; + let was_inserted = inserted_received_at.is_some(); + // On a conflict (duplicate/replay) no row is returned; callers gate on + // `was_inserted`, and the resurface replay path re-reads the original + // event's `received_at`. + let received_at = inserted_received_at.unwrap_or_else(Utc::now); if was_inserted { if let Some(ref meta) = thread_meta { @@ -1656,6 +1671,57 @@ mod tests { assert_eq!(persisted, 0); } + /// The DM resurface fence (`unhide_dm_recipients`) compares an event's + /// `received_at` against `channel_members.hidden_at`, which is written with + /// the database clock (`hide_dm` uses `NOW()`). For that comparison to be + /// causally sound under relay/DB clock skew, `received_at` must also come + /// from the database clock — not the relay process `Utc::now()`. + /// + /// This asserts the persisted `received_at` falls inside a `[before, after]` + /// window sampled from the *same database's* `NOW()`, and that the value the + /// insert returns matches the persisted column. A relay-process timestamp + /// would only satisfy this by accident of zero skew. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn insert_event_sources_received_at_from_the_database_clock() { + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let event = make_text_event("db-clock received_at"); + + let before: DateTime = sqlx::query_scalar("SELECT NOW()") + .fetch_one(&pool) + .await + .expect("read db NOW() before insert"); + let (stored, was_inserted) = insert_event(&pool, community, &event, None) + .await + .expect("insert event"); + let after: DateTime = sqlx::query_scalar("SELECT NOW()") + .fetch_one(&pool) + .await + .expect("read db NOW() after insert"); + + assert!(was_inserted, "fresh event must insert"); + assert!( + stored.received_at >= before && stored.received_at <= after, + "received_at {} must fall within the database clock window [{before}, {after}]", + stored.received_at + ); + + let persisted: DateTime = sqlx::query_scalar( + "SELECT received_at FROM events WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("read persisted received_at"); + assert_eq!( + stored.received_at, persisted, + "the returned received_at must equal the persisted database value" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn event_insert_ttl_trigger_handles_permanent_ephemeral_duplicate_and_activation_race() { diff --git a/desktop/src-tauri/src/commands/channels/fetch.rs b/desktop/src-tauri/src/commands/channels/fetch.rs index 36c24a35b7d..249e5af5b1e 100644 --- a/desktop/src-tauri/src/commands/channels/fetch.rs +++ b/desktop/src-tauri/src/commands/channels/fetch.rs @@ -289,8 +289,12 @@ pub(super) async fn fetch_channels( DirectoryScope::MemberOnly => Ok(Vec::new()), } }, - // Step 6: NIP-DV hidden-DM snapshot. Tolerant — a failure means no DMs - // are hidden rather than aborting the whole fetch. + // Step 6: NIP-DV hidden-DM snapshot. A read *failure* must NOT be + // treated as "no DMs are hidden": that fail-open collapse would + // re-expose every hidden DM until the next successful refresh. Only a + // successful read of zero snapshots means nothing is hidden. Propagate + // the transport error so the whole refresh fails and the caller keeps + // the previously rendered (correctly filtered) channel list. async { let events = query_relay( state, @@ -300,21 +304,22 @@ pub(super) async fn fetch_channels( "limit": 1, })], ) - .await - .unwrap_or_default(); - events - .iter() - .max_by_key(|e| e.created_at.as_secs()) - .map(|e| { - e.tags - .iter() - .filter_map(|t| { - let s = t.as_slice(); - (s.len() >= 2 && s[0] == "h").then(|| s[1].clone()) - }) - .collect::>() - }) - .unwrap_or_default() + .await?; + Ok::, String>( + events + .iter() + .max_by_key(|e| e.created_at.as_secs()) + .map(|e| { + e.tags + .iter() + .filter_map(|t| { + let s = t.as_slice(); + (s.len() >= 2 && s[0] == "h").then(|| s[1].clone()) + }) + .collect::>() + }) + .unwrap_or_default(), + ) }, ); @@ -323,7 +328,9 @@ pub(super) async fn fetch_channels( let meta_events = member_chain_result?; let open_meta_events = open_meta_result?; - // hidden_dms is already a resolved HashSet (tolerant path above) + // A snapshot read failure fails the whole refresh (fail-closed): the caller + // preserves the last rendered list rather than showing every hidden DM. + let hidden_dms = hidden_dms?; // Merge: member channels (marked as member) + non-member channels (open // directory when included, else pending-owned) not already in the member set. diff --git a/desktop/src/features/channels/useMembershipNotifications.test.mjs b/desktop/src/features/channels/useMembershipNotifications.test.mjs index 8ebfa550132..bc1c11255d8 100644 --- a/desktop/src/features/channels/useMembershipNotifications.test.mjs +++ b/desktop/src/features/channels/useMembershipNotifications.test.mjs @@ -48,6 +48,10 @@ test("DM visibility has a dedicated replay and refreshes only the channel list", queryClient.invalidateQueries = async ({ queryKey }) => { invalidations.push(queryKey); }; + // The channel-list invalidation is routed through an idle-aware trailing + // debounce; report the query idle so a fired debounce invalidates rather than + // re-arms. Detail/members keys are invalidated directly. + queryClient.isFetching = () => 0; const wrapper = ({ children }) => React.createElement(QueryClientProvider, { client: queryClient }, children); @@ -91,6 +95,10 @@ test("DM visibility has a dedicated replay and refreshes only the channel list", sig: "sig", }); }); + // The channel-list refresh is debounced (trailing). Let the timer fire. + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 600)); + }); assert.deepEqual(invalidations, [["channels"]]); invalidations.length = 0; @@ -108,10 +116,18 @@ test("DM visibility has a dedicated replay and refreshes only the channel list", sig: "sig", }); }); + // Detail/members invalidate synchronously; the channel-list refresh trails. + assert.deepEqual(invalidations, [ + ["channels", "new-channel", "detail"], + ["channels", "new-channel", "members"], + ]); + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 600)); + }); assert.deepEqual(invalidations, [ - ["channels"], ["channels", "new-channel", "detail"], ["channels", "new-channel", "members"], + ["channels"], ]); unmount(); @@ -121,3 +137,84 @@ test("DM visibility has a dedicated replay and refreshes only the channel list", relayClient.subscribeLive = originalSubscribeLive; } }); + +test("channel-list refresh re-arms instead of dropping the signal mid-fetch", async () => { + const React = await import("react"); + const { act, cleanup, renderHook } = await import("@testing-library/react"); + const { QueryClient, QueryClientProvider } = await import( + "@tanstack/react-query" + ); + const { relayClient } = await import("@/shared/api/relayClient"); + const { KIND_DM_VISIBILITY } = await import("@/shared/constants/kinds"); + const { useMembershipNotifications } = await import( + "./useMembershipNotifications.ts" + ); + + const originalSubscribeLive = relayClient.subscribeLive; + const subscriptions = []; + relayClient.subscribeLive = async (filter, listener) => { + subscriptions.push({ filter, listener }); + return async () => {}; + }; + + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const invalidations = []; + queryClient.invalidateQueries = async ({ queryKey }) => { + invalidations.push(queryKey); + }; + // Simulate get_channels in flight: a direct invalidate now would be silently + // undone when the older (pre-event) response lands. The idle-aware routing + // must re-arm instead, so no invalidation lands while fetching, and exactly + // one lands once the query goes idle. + let fetching = 1; + queryClient.isFetching = () => fetching; + const wrapper = ({ children }) => + React.createElement(QueryClientProvider, { client: queryClient }, children); + + try { + renderHook(() => useMembershipNotifications("viewer-pubkey"), { wrapper }); + await act(async () => new Promise((resolve) => setImmediate(resolve))); + + const visibility = subscriptions.find( + (subscription) => + subscription.filter.kinds.length === 1 && + subscription.filter.kinds[0] === KIND_DM_VISIBILITY, + ); + + await act(async () => { + visibility.listener({ + id: "visibility", + pubkey: "relay", + created_at: 1, + kind: KIND_DM_VISIBILITY, + tags: [["p", "viewer-pubkey"]], + content: "", + sig: "sig", + }); + }); + // Debounce fires while a fetch is in flight -> must re-arm, not invalidate. + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 600)); + }); + assert.deepEqual( + invalidations, + [], + "must not invalidate while get_channels is in flight", + ); + + // Query goes idle; the re-armed refresh now invalidates exactly once. + fetching = 0; + await act(async () => { + await new Promise((resolve) => setTimeout(resolve, 600)); + }); + assert.deepEqual(invalidations, [["channels"]]); + + cleanup(); + } finally { + cleanup(); + queryClient.clear(); + relayClient.subscribeLive = originalSubscribeLive; + } +}); diff --git a/desktop/src/features/channels/useMembershipNotifications.ts b/desktop/src/features/channels/useMembershipNotifications.ts index f66e672b71c..371c2f781a5 100644 --- a/desktop/src/features/channels/useMembershipNotifications.ts +++ b/desktop/src/features/channels/useMembershipNotifications.ts @@ -2,9 +2,14 @@ import * as React from "react"; import { useQueryClient } from "@tanstack/react-query"; import { channelsQueryKey } from "@/features/channels/hooks"; +import { refreshChannelsWhenIdle } from "@/features/channels/refreshChannelsWhenIdle"; import { getChannelIdFromTags } from "@/features/messages/lib/threading"; import { relayClient } from "@/shared/api/relayClient"; import type { RelayEvent } from "@/shared/api/types"; +import { + createTrailingDebounce, + type TrailingDebounce, +} from "@/shared/lib/trailingDebounce"; import { KIND_DM_VISIBILITY, KIND_MEMBER_ADDED_NOTIFICATION, @@ -13,16 +18,39 @@ import { const MEMBERSHIP_NOTIFICATION_RETRY_BASE_MS = 1_000; const MEMBERSHIP_NOTIFICATION_RETRY_MAX_MS = 30_000; +// Matches useLiveChannelUpdates: collapse a burst of notifications into a +// single trailing refresh once the channels query goes idle. +const CHANNELS_INVALIDATE_DEBOUNCE_MS = 500; export function useMembershipNotifications(currentPubkey?: string) { const queryClient = useQueryClient(); const normalizedCurrentPubkey = currentPubkey?.trim().toLowerCase() ?? ""; + // Route channel-list invalidation through the idle-aware trailing mechanism + // rather than invalidating directly. A direct invalidate while get_channels + // is mid-flight is silently undone when the older (pre-event) response lands + // and clears the dirty flag — dropping the resurface signal so an incoming + // DM stays out of navigation until the next poll. This mirrors ordinary live + // channel traffic (useLiveChannelUpdates), which invalidates the same key. + const channelsInvalidateRef = React.useRef(null); + if (channelsInvalidateRef.current === null) { + channelsInvalidateRef.current = createTrailingDebounce(() => { + refreshChannelsWhenIdle({ + isFetching: () => + queryClient.isFetching({ queryKey: channelsQueryKey }), + invalidate: () => { + void queryClient.invalidateQueries({ queryKey: channelsQueryKey }); + }, + reArm: () => channelsInvalidateRef.current?.trigger(), + }); + }, CHANNELS_INVALIDATE_DEBOUNCE_MS); + } + const handleMembershipNotification = React.useEffectEvent( (event: RelayEvent) => { const channelId = getChannelIdFromTags(event.tags); - void queryClient.invalidateQueries({ queryKey: channelsQueryKey }); + channelsInvalidateRef.current?.trigger(); if (event.kind === KIND_DM_VISIBILITY) { return; } @@ -134,6 +162,7 @@ export function useMembershipNotifications(currentPubkey?: string) { if (retryTimeout !== undefined) { window.clearTimeout(retryTimeout); } + channelsInvalidateRef.current?.cancel(); if (dispose) { void dispose().catch(() => {}); } diff --git a/mobile/lib/features/channels/channel_directory.dart b/mobile/lib/features/channels/channel_directory.dart index a99822ede5e..db11b14a1ed 100644 --- a/mobile/lib/features/channels/channel_directory.dart +++ b/mobile/lib/features/channels/channel_directory.dart @@ -225,24 +225,29 @@ Future> _resolveDmDisplayNames( return displayNames; } +/// Fetch the set of DM channel ids the viewer currently has hidden, from the +/// relay-signed NIP-DV snapshot. +/// +/// A read *failure* is propagated to the caller rather than swallowed: treating +/// a transient history-read error as "nothing is hidden" would fail open and +/// re-expose every hidden DM until the next successful refresh. An empty result +/// (`events.isEmpty`) is a legitimate "no DMs hidden" and returns `{}`. Callers +/// wrap this in `_fenced` and preserve the prior list on failure, matching the +/// sibling membership/huddle reads. Future> _fetchHiddenDmIds( RelaySessionNotifier session, String myPk, ) async { - try { - final events = await session.fetchHistory(NostrFilters.hiddenDms(myPk)); - if (events.isEmpty) return const {}; - NostrEvent latest = events.first; - for (final event in events.skip(1)) { - if (event.createdAt > latest.createdAt) latest = event; - } - return { - for (final tag in latest.tags) - if (tag.length >= 2 && tag[0] == 'h') tag[1], - }; - } catch (_) { - return const {}; + final events = await session.fetchHistory(NostrFilters.hiddenDms(myPk)); + if (events.isEmpty) return const {}; + NostrEvent latest = events.first; + for (final event in events.skip(1)) { + if (event.createdAt > latest.createdAt) latest = event; } + return { + for (final tag in latest.tags) + if (tag.length >= 2 && tag[0] == 'h') tag[1], + }; } Future> _fetchHuddleStarts( diff --git a/mobile/lib/features/channels/channels_provider/dm_visibility.dart b/mobile/lib/features/channels/channels_provider/dm_visibility.dart index 1820027666e..a9ab412a438 100644 --- a/mobile/lib/features/channels/channels_provider/dm_visibility.dart +++ b/mobile/lib/features/channels/channels_provider/dm_visibility.dart @@ -4,6 +4,16 @@ mixin _DmVisibilitySubscription on AsyncNotifier> { void Function()? _unsubscribeDmVisibility; String? _dmVisibilityRelayBaseUrl; String? _dmVisibilityPubkey; + // Id of the last visibility snapshot we acted on. The `limit: 1` filter + // replays the latest stored kind:30622 snapshot on every (re)subscribe; if + // that replay fires before `subscribe()` resolves, `refresh()` bumps the + // subscription version, retires this install, and the next queued sync + // re-subscribes and replays the identical snapshot again — an unbounded + // refresh/subscribe loop for any user with an existing snapshot. Snapshots + // are parameterized-replaceable, so the stored event keeps a stable id; + // suppressing a re-delivered identical id breaks that loop while a genuinely + // new snapshot (new id) still triggers exactly one refresh. + String? _lastHandledDmVisibilityEventId; int get _subscriptionVersion; @@ -22,6 +32,9 @@ mixin _DmVisibilitySubscription on AsyncNotifier> { _unsubscribeDmVisibility = null; _dmVisibilityRelayBaseUrl = relayBaseUrl; _dmVisibilityPubkey = myPk; + // A different relay/identity owns entirely different snapshots; drop the + // dedup key so the new scope's first snapshot is honored. + _lastHandledDmVisibilityEventId = null; } if (_unsubscribeDmVisibility != null || myPk == null) return; @@ -61,6 +74,11 @@ mixin _DmVisibilitySubscription on AsyncNotifier> { void _handleDmVisibilityEvent(NostrEvent event) { if (event.kind != EventKind.dmVisibility) return; + // Suppress a re-delivered identical snapshot (see `_lastHandledDmVisibilityEventId`). + // The relay replays the latest snapshot on every resubscribe; only a new + // snapshot id should drive a refresh. + if (event.id == _lastHandledDmVisibilityEventId) return; + _lastHandledDmVisibilityEventId = event.id; unawaited(refresh()); } @@ -69,6 +87,7 @@ mixin _DmVisibilitySubscription on AsyncNotifier> { _unsubscribeDmVisibility = null; _dmVisibilityRelayBaseUrl = null; _dmVisibilityPubkey = null; + _lastHandledDmVisibilityEventId = null; } Set _mutedChannelIds() => { diff --git a/mobile/test/features/channels/channels_provider_test.dart b/mobile/test/features/channels/channels_provider_test.dart index cd65db5a421..4054e1d0522 100644 --- a/mobile/test/features/channels/channels_provider_test.dart +++ b/mobile/test/features/channels/channels_provider_test.dart @@ -2085,6 +2085,42 @@ void main() { expect(session.visibilitySubscribeFilters.single.tags['#p'], [myPk]); }); + test('a failed hidden-DM read keeps DMs hidden (fail-closed)', () async { + // A transient hidden-DM snapshot read failure must NOT collapse to "no + // DMs are hidden": that fail-open behavior would re-expose every hidden + // DM until the next successful refresh. The read now propagates its error + // like the sibling membership/huddle reads, so a refresh that can't read + // the snapshot fails and preserves the previously rendered (correctly + // filtered) list instead of surfacing the hidden DM. + final session = _FakeRelaySession( + memberships: [_membership(_channelA, myPk), _membership(_channelB, myPk)], + metadata: [ + _meta(id: _channelA, name: 'Alice', channelType: 'dm'), + _meta(id: _channelB, name: 'Bob', channelType: 'dm'), + ], + hiddenDmEvents: [ + _hiddenDms([_channelA], pubkey: myPk), + ], + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + + final initial = await container.read(channelsProvider.future); + expect(initial.map((c) => c.id), [_channelB]); + + // The next snapshot read fails. `retryDirectory` preserves the prior list + // on any refresh failure; the hidden DM must stay hidden, never resurface. + session.hiddenDmFailures = 1; + await container.read(channelsProvider.notifier).retryDirectory(); + + expect( + container.read(channelsProvider).requireValue.map((c) => c.id), + [_channelB], + reason: + 'a hidden-DM read failure must not re-expose the hidden DM (fail-closed)', + ); + }); + test('a DM visibility snapshot refreshes the hidden channel list', () async { final hiddenDmEvents = [ _hiddenDms([_channelA], pubkey: myPk), @@ -2131,6 +2167,85 @@ void main() { ); }); + test( + 'a re-delivered identical visibility snapshot does not re-refresh', + () async { + // The `limit: 1` DM-visibility subscription replays the latest stored + // snapshot on every (re)subscribe. If a replay that repeats the snapshot + // we already acted on triggered another refresh, the refresh would bump + // the subscription version, discard the in-flight subscription, resubscribe, + // and replay the same snapshot again — an unbounded refresh/resubscribe + // loop. Snapshots are parameterized-replaceable (stable id), so a repeated + // id must be suppressed while a genuinely new snapshot still refreshes. + final hiddenDmEvents = [ + _hiddenDms([_channelA], pubkey: myPk), + ]; + final session = _FakeRelaySession( + memberships: [ + _membership(_channelA, myPk), + _membership(_channelB, myPk), + ], + metadata: [ + _meta(id: _channelA, name: 'Alice', channelType: 'dm'), + _meta(id: _channelB, name: 'Bob', channelType: 'dm'), + ], + hiddenDmEvents: hiddenDmEvents, + ); + final container = _buildContainer(session: session); + addTearDown(container.dispose); + + expect( + (await container.read( + channelsProvider.future, + )).map((channel) => channel.id), + [_channelB], + ); + final refreshesAfterLoad = session.membershipRequestCount; + + // A new snapshot (channelA no longer hidden) drives exactly one refresh. + final unhide = _hiddenDms(const [], pubkey: myPk); + hiddenDmEvents + ..clear() + ..add(unhide); + session.emit(unhide); + await _waitUntil( + () => + container + .read(channelsProvider) + .value + ?.map((channel) => channel.id) + .toSet() + .length == + 2, + ); + final refreshesAfterNewSnapshot = session.membershipRequestCount; + expect( + refreshesAfterNewSnapshot, + greaterThan(refreshesAfterLoad), + reason: 'a new snapshot id must refresh', + ); + + // Re-delivering the SAME snapshot id (the replay) must NOT refresh again. + session.emit(unhide); + session.emit(unhide); + // Let any erroneously-scheduled refresh run. + await Future.delayed(const Duration(milliseconds: 50)); + expect( + session.membershipRequestCount, + refreshesAfterNewSnapshot, + reason: + 'a repeated identical snapshot must be suppressed, not re-refreshed', + ); + expect( + container + .read(channelsProvider) + .requireValue + .map((channel) => channel.id), + unorderedEquals([_channelA, _channelB]), + ); + }, + ); + test( 'archived kind:39000 metadata sets Channel.isArchived (covers TTL auto-archive)', () async { @@ -2489,6 +2604,7 @@ class _FakeRelaySession extends RelaySessionNotifier { List recentMessages; int membershipFailures; int directoryFailures = 0; + int hiddenDmFailures = 0; bool failClaimedMemberCountQuery = false; bool failClaimedUnreadCatchUpQuery = false; int membershipRequestCount = 0; @@ -2742,6 +2858,10 @@ class _FakeRelaySession extends RelaySessionNotifier { _hiddenDmStarted = null; await paused.future; } + if (hiddenDmFailures > 0) { + hiddenDmFailures--; + throw Exception('hidden-DM snapshot fetch failed'); + } return hiddenDmEvents; } if (filter.kinds.contains(EventKind.huddleStarted)) {