diff --git a/desktop/src-tauri/src/commands/channels.rs b/desktop/src-tauri/src/commands/channels.rs index fd74b3d893..54a6037c23 100644 --- a/desktop/src-tauri/src/commands/channels.rs +++ b/desktop/src-tauri/src/commands/channels.rs @@ -146,15 +146,15 @@ fn profile_join_pubkeys(members: &[crate::models::ChannelMemberInfo], limit: usi #[tauri::command] pub async fn get_channel_members( channel_id: String, + read_your_writes: Option, state: State<'_, AppState>, ) -> Result { let events = query_relay( &state, - &[serde_json::json!({ - "kinds": [39002], - "#d": [channel_id], - "limit": 1 - })], + &[channel_members_filter( + &channel_id, + read_your_writes.unwrap_or(false), + )], ) .await?; @@ -249,6 +249,7 @@ async fn ensure_starter_channel_memberships( state: &AppState, keys: &nostr::Keys, channels: &mut [ChannelInfo], + changed_channel_ids: &mut Vec, ) -> Result<(), String> { for spec in STARTER_CHANNELS { let Some(channel) = channels @@ -266,11 +267,41 @@ async fn ensure_starter_channel_memberships( let builder = events::build_join(channel_uuid)?; submit_event_with_keys(builder, state, keys, None).await?; channel.is_member = true; + changed_channel_ids.push(channel.id.clone()); } Ok(()) } +/// Build the kind:39002 member-list filter. Display reads stay replica-eligible; +/// `read_your_writes` is for callers acting on state this client just wrote. +fn channel_members_filter(channel_id: &str, read_your_writes: bool) -> serde_json::Value { + let mut filter = serde_json::json!({ + "kinds": [39002], + "#d": [channel_id], + "limit": 1, + }); + if read_your_writes { + // Read-your-writes: pin to the writer so a lagging replica cannot hide + // the member list the relay wrote for a just-created channel. + filter["consistency"] = serde_json::json!("strong"); + } + filter +} + +/// Build the kind:39000 filter that reads back channel metadata by ID right +/// after this client created or edited those channels. +pub(crate) fn channel_metadata_filter(channel_ids: &[impl serde::Serialize]) -> serde_json::Value { + serde_json::json!({ + "kinds": [39000], + "#d": channel_ids, + "limit": channel_ids.len(), + // Read-your-writes: a lagging read replica would miss the metadata the + // relay just wrote, so pin this read to the writer. + "consistency": "strong", + }) +} + async fn fetch_starter_channel_metadata( state: &AppState, channel_ids: &[String], @@ -279,15 +310,7 @@ async fn fetch_starter_channel_metadata( return Ok(Vec::new()); } - let events = query_relay( - state, - &[serde_json::json!({ - "kinds": [39000], - "#d": channel_ids, - "limit": channel_ids.len(), - })], - ) - .await?; + let events = query_relay(state, &[channel_metadata_filter(channel_ids)]).await?; events .iter() @@ -343,15 +366,7 @@ pub async fn create_channel( state.mark_pending_owned_channel(&creator_pubkey, &channel_uuid_string); // Re-fetch the canonical metadata event to return ChannelInfo. - let events = query_relay( - &state, - &[serde_json::json!({ - "kinds": [39000], - "#d": [channel_uuid_string], - "limit": 1 - })], - ) - .await?; + let events = query_relay(&state, &[channel_metadata_filter(&[&channel_uuid_string])]).await?; events .first() @@ -360,13 +375,40 @@ pub async fn create_channel( .ok_or_else(|| "channel created but metadata not yet available".to_string()) } +/// Starter channels plus every channel this call created or joined. Any +/// failure is reported in `error` rather than as the command error, so the +/// writes accepted before it are still reported; the frontend rethrows +/// `error` unchanged. +#[derive(serde::Serialize)] +pub struct StarterChannelsResult { + channels: Vec, + changed_channel_ids: Vec, + error: Option, +} + #[tauri::command] pub async fn ensure_starter_channels( state: State<'_, AppState>, +) -> Result { + let mut changed_channel_ids = Vec::new(); + let (channels, error) = + match ensure_starter_channels_inner(&state, &mut changed_channel_ids).await { + Ok(channels) => (channels, None), + Err(error) => (Vec::new(), Some(error)), + }; + Ok(StarterChannelsResult { + channels, + changed_channel_ids, + error, + }) +} + +async fn ensure_starter_channels_inner( + state: &AppState, + changed_channel_ids: &mut Vec, ) -> Result, String> { - let mut existing_channels = - fetch_channels(&state, DirectoryScope::IncludeOpenDirectory).await?; - let relay_scope = relay_api_base_url_with_override(&state); + let mut existing_channels = fetch_channels(state, DirectoryScope::IncludeOpenDirectory).await?; + let relay_scope = relay_api_base_url_with_override(state); let creator_keys = state.signing_keys()?; let creator_pubkey = creator_keys.public_key().to_hex(); let mut starter_ids = Vec::with_capacity(STARTER_CHANNELS.len()); @@ -392,10 +434,11 @@ pub async fn ensure_starter_channels( None, )?; - match submit_event_with_keys(builder, &state, &creator_keys, None).await { + match submit_event_with_keys(builder, state, &creator_keys, None).await { Ok(_) => { state.mark_pending_owned_channel(&creator_pubkey, &channel_uuid_string); created_ids.insert(channel_uuid_string.clone()); + changed_channel_ids.push(channel_uuid_string.clone()); } Err(error) if is_duplicate_channel_rejection(&error) => { state.mark_pending_owned_channel(&creator_pubkey, &channel_uuid_string); @@ -405,7 +448,7 @@ pub async fn ensure_starter_channels( } for _ in 0..3 { - let metadata = fetch_starter_channel_metadata(&state, &starter_ids).await?; + let metadata = fetch_starter_channel_metadata(state, &starter_ids).await?; for mut channel in metadata { if created_ids.contains(&channel.id) { channel.is_member = true; @@ -424,14 +467,20 @@ pub async fn ensure_starter_channels( } if !has_all_starter_channels(&existing_channels) { - existing_channels = fetch_channels(&state, DirectoryScope::IncludeOpenDirectory).await?; + existing_channels = fetch_channels(state, DirectoryScope::IncludeOpenDirectory).await?; } if !has_all_starter_channels(&existing_channels) { return Err("starter channels created but metadata not yet available".to_string()); } - ensure_starter_channel_memberships(&state, &creator_keys, &mut existing_channels).await?; + ensure_starter_channel_memberships( + state, + &creator_keys, + &mut existing_channels, + changed_channel_ids, + ) + .await?; Ok(existing_channels) } @@ -465,15 +514,7 @@ pub async fn update_channel( )?; submit_event(builder, &state).await?; - let events = query_relay( - &state, - &[serde_json::json!({ - "kinds": [39000], - "#d": [input.channel_id], - "limit": 1 - })], - ) - .await?; + let events = query_relay(&state, &[channel_metadata_filter(&[&input.channel_id])]).await?; events .first() @@ -629,3 +670,6 @@ pub async fn leave_channel(channel_id: String, state: State<'_, AppState>) -> Re #[cfg(test)] #[path = "channels_tests.rs"] mod tests; + +#[cfg(test)] +mod starter_tests; diff --git a/desktop/src-tauri/src/commands/channels/starter_tests.rs b/desktop/src-tauri/src/commands/channels/starter_tests.rs new file mode 100644 index 0000000000..71bbbd0e89 --- /dev/null +++ b/desktop/src-tauri/src/commands/channels/starter_tests.rs @@ -0,0 +1,40 @@ +//! Starter-channel creation must report every channel it created, even when a +//! later step fails, so the frontend can route that channel's roster reads to +//! the writer. +use super::*; +use axum::{routing::post, Json, Router}; +use nostr::Keys; +use serde_json::json; + +#[tokio::test] +async fn created_starter_channels_are_reported_when_metadata_is_unavailable() { + let _serial = crate::relay_admission::TEST_SERIAL.lock().await; + // The relay accepts every create, but its replica never serves metadata. + let router = Router::new() + .route("/query", post(|| async { Json(json!([])) })) + .route( + "/events", + post(|| async { Json(json!({"event_id": "0", "accepted": true, "message": ""})) }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + let state = crate::app_state::build_app_state(); + *state.keys.lock().unwrap() = Keys::generate(); + *state.relay_url_override.lock().unwrap() = Some(url); + + let mut changed = Vec::new(); + let result = ensure_starter_channels_inner(&state, &mut changed).await; + server.abort(); + + assert_eq!( + result.err().as_deref(), + Some("starter channels created but metadata not yet available") + ); + let scope = relay_api_base_url_with_override(&state); + let expected: Vec = STARTER_CHANNELS + .iter() + .map(|spec| starter_channel_uuid(&scope, spec.slug).to_string()) + .collect(); + assert_eq!(changed, expected); +} diff --git a/desktop/src-tauri/src/commands/channels_tests.rs b/desktop/src-tauri/src/commands/channels_tests.rs index 91e636d5f2..d31acb08f8 100644 --- a/desktop/src-tauri/src/commands/channels_tests.rs +++ b/desktop/src-tauri/src/commands/channels_tests.rs @@ -513,3 +513,39 @@ fn profile_join_pubkeys_caps_in_roster_order() { assert_eq!(profile_join_pubkeys(&members, 10).len(), 3); assert!(profile_join_pubkeys(&[], 10).is_empty()); } + +// Mutation oracle: removing `"consistency"` from `channel_metadata_filter` +// fails this test, and create/update/starter/open-DM all read back through it. +#[test] +fn channel_metadata_filter_carries_strong_consistency() { + let ids = ["a", "b"]; + let f = channel_metadata_filter(&ids); + assert_eq!( + f.get("consistency").and_then(|v| v.as_str()), + Some("strong"), + "channel metadata read-back must carry consistency=strong: {f}" + ); + assert_eq!(f["kinds"], serde_json::json!([39000])); + assert_eq!(f["#d"], serde_json::json!(["a", "b"])); + assert_eq!(f.get("limit").and_then(|v| v.as_u64()), Some(2)); +} + +// Mutation oracle: dropping `"consistency"` from the read-your-writes arm of +// `channel_members_filter`, or adding it to the default arm, fails this test. +#[test] +fn channel_members_filter_pins_writer_only_for_read_your_writes() { + let pinned = channel_members_filter("a", true); + assert_eq!( + pinned.get("consistency").and_then(|v| v.as_str()), + Some("strong"), + "read-your-writes member read must carry consistency=strong: {pinned}" + ); + assert_eq!(pinned["kinds"], serde_json::json!([39002])); + assert_eq!(pinned["#d"], serde_json::json!(["a"])); + + let display = channel_members_filter("a", false); + assert!( + display.get("consistency").is_none(), + "display member read must stay replica-eligible: {display}" + ); +} diff --git a/desktop/src-tauri/src/commands/dms.rs b/desktop/src-tauri/src/commands/dms.rs index 252068c94d..e916c2805a 100644 --- a/desktop/src-tauri/src/commands/dms.rs +++ b/desktop/src-tauri/src/commands/dms.rs @@ -3,6 +3,7 @@ use tauri::State; use crate::{ app_state::AppState, + commands::channels::channel_metadata_filter, events, models::ChannelInfo, nostr_convert, @@ -66,11 +67,7 @@ pub(crate) async fn open_dm_with_scope( let metadata = query_relay_at_with_keys( state, &api_base_url, - &[serde_json::json!({ - "kinds": [39000], - "#d": [ack.channel_id], - "limit": 1 - })], + &[channel_metadata_filter(&[&ack.channel_id])], &keys, None, ) diff --git a/desktop/src-tauri/src/egress_guard_tests.rs b/desktop/src-tauri/src/egress_guard_tests.rs index 4e2c5c4132..5511aa8d72 100644 --- a/desktop/src-tauri/src/egress_guard_tests.rs +++ b/desktop/src-tauri/src/egress_guard_tests.rs @@ -293,6 +293,8 @@ const EVENTS_INVENTORY: &[(&str, usize, usize)] = &[ // Stub-relay route in the tombstone-flush gate tests; production flush // publishes through the guarded boundary-1 funnel. ("src/commands/teams/pending/tests/gate.rs", 1, 0), + // Accepting stub relay in the starter-channel creation tests. + ("src/commands/channels/starter_tests.rs", 1, 0), // Admin API mutations: every JSON body passes the guard in // `helpers::send_admin_mutation` (injection tests in admin `direct_action_tests.rs`). // The `/events/{id}/delete` route + its path test and the native delete diff --git a/desktop/src-tauri/src/huddle/agents.rs b/desktop/src-tauri/src/huddle/agents.rs index 2bdf054426..3213badf89 100644 --- a/desktop/src-tauri/src/huddle/agents.rs +++ b/desktop/src-tauri/src/huddle/agents.rs @@ -75,6 +75,12 @@ pub struct AgentHuddleSyncResult { pub matched_active_huddle: bool, /// Agents newly enrolled in the Huddle's ephemeral channel. pub added: Vec, + /// Channels whose membership this sync changed (ephemeral, and the parent + /// when an agent was enrolled there), so the frontend can refresh them. + /// Reported even when `error` is set, because earlier adds still landed. + pub changed_channel_ids: Vec, + /// Failure that stopped the sync after `changed_channel_ids` were written. + pub error: Option, } // Multiple frontend mutation paths can observe the same membership addition @@ -163,6 +169,8 @@ pub(crate) async fn sync_agents_for_active_huddle( return Ok(AgentHuddleSyncResult { matched_active_huddle: false, added: Vec::new(), + changed_channel_ids: Vec::new(), + error: None, }); } let _sync_guard = AGENT_SYNC_LOCK.lock().await; @@ -173,6 +181,8 @@ pub(crate) async fn sync_agents_for_active_huddle( return Ok(AgentHuddleSyncResult { matched_active_huddle: false, added: Vec::new(), + changed_channel_ids: Vec::new(), + error: None, }); } let ephemeral_channel_id = huddle @@ -187,6 +197,8 @@ pub(crate) async fn sync_agents_for_active_huddle( return Ok(AgentHuddleSyncResult { matched_active_huddle: false, added: Vec::new(), + changed_channel_ids: Vec::new(), + error: None, }); } let state_agents = huddle @@ -231,8 +243,26 @@ pub(crate) async fn sync_agents_for_active_huddle( let ephemeral_uuid = Uuid::parse_str(&ephemeral_channel_id).map_err(|e| e.to_string())?; let parent_uuid = Uuid::parse_str(&parent_channel_id).map_err(|e| e.to_string())?; let mut added = Vec::new(); + let mut changed_channel_ids = Vec::new(); for pubkey in missing { - add_agent_to_huddle(ephemeral_uuid, parent_uuid, &pubkey, state).await?; + match add_agent_to_huddle(ephemeral_uuid, parent_uuid, &pubkey, state).await { + Ok(result) => { + note_changed(&mut changed_channel_ids, &ephemeral_channel_id); + if result.parent_added { + note_changed(&mut changed_channel_ids, &parent_channel_id); + } + } + Err(error) => { + // The ephemeral add may have landed before the failure. + note_changed(&mut changed_channel_ids, &ephemeral_channel_id); + return Ok(AgentHuddleSyncResult { + matched_active_huddle: true, + added, + changed_channel_ids, + error: Some(error), + }); + } + } merged_agents.push(pubkey.clone()); added.push(pubkey); } @@ -243,6 +273,8 @@ pub(crate) async fn sync_agents_for_active_huddle( return Ok(AgentHuddleSyncResult { matched_active_huddle: true, added, + changed_channel_ids, + error: None, }); } let mut roster_changed = false; @@ -277,6 +309,8 @@ pub(crate) async fn sync_agents_for_active_huddle( Ok(AgentHuddleSyncResult { matched_active_huddle: true, added, + changed_channel_ids, + error: None, }) } @@ -289,6 +323,12 @@ pub async fn sync_agents_to_active_huddle( sync_agents_for_active_huddle(&channel_id, agent_pubkeys, &state).await } +fn note_changed(changed: &mut Vec, channel_id: &str) { + if !changed.iter().any(|id| id == channel_id) { + changed.push(channel_id.to_owned()); + } +} + fn contains_member(members: &[(String, Option)], pubkey: &str) -> bool { members .iter() diff --git a/desktop/src/app/App.tsx b/desktop/src/app/App.tsx index a05b66f016..581987ed61 100644 --- a/desktop/src/app/App.tsx +++ b/desktop/src/app/App.tsx @@ -60,6 +60,7 @@ import { WelcomeSetup } from "@/features/communities/ui/WelcomeSetup"; import { CommunityApplyErrorScreen } from "@/features/communities/ui/CommunityApplyErrorScreen"; import { CommunityChangeOverlay } from "@/features/communities/ui/CommunityChangeOverlay"; import { setAvatarProfileSyncQueryClient } from "@/features/profile/avatarProfileSync"; +import { refreshRostersOnMembershipChange } from "@/features/channels/rosterFreshness"; import { seedProjectSnapshot } from "@/features/projects/projectSnapshot"; import { EncryptedBackupProvider } from "@/features/settings/EncryptedBackupProvider"; import { createBuzzQueryClient } from "@/shared/api/queryClient"; @@ -241,6 +242,7 @@ function CommunityQueryProvider({ }); useEffect(() => setAvatarProfileSyncQueryClient(queryClient), [queryClient]); + useEffect(() => refreshRostersOnMembershipChange(queryClient), [queryClient]); useEffect(() => { const e2eWindow = window as Window & { diff --git a/desktop/src/features/admin-console/AdminConsoleActionsTab.tsx b/desktop/src/features/admin-console/AdminConsoleActionsTab.tsx index 4bd32d1883..c8299e8209 100644 --- a/desktop/src/features/admin-console/AdminConsoleActionsTab.tsx +++ b/desktop/src/features/admin-console/AdminConsoleActionsTab.tsx @@ -35,6 +35,8 @@ import { directAdminAction, type AdminDirectAction, type AdminDirectIntent, + communityHostFromRelayUrl, + normalizeCommunityHost, } from "./api"; import { adminErrorCode, @@ -79,30 +81,6 @@ function directErrorMessage(e: unknown): string { } } -/** - * Normalize a community host the way the relay's `normalize_host` does: - * lowercase, no default port, no trailing root dot. - */ -function normalizeCommunityHost(raw: string): string { - return raw - .trim() - .toLowerCase() - .replace(/:(443|80)$/, "") - .replace(/\.$/, ""); -} - -/** - * The community a relay URL serves is its `Host`: the URL's authority, with - * the default port already dropped for `ws`/`wss`. Null when unparseable. - */ -export function communityHostFromRelayUrl(relayUrl: string): string | null { - try { - return normalizeCommunityHost(new URL(relayUrl).host) || null; - } catch { - return null; - } -} - /** An `nsec` or NIP-49 backup, in either case. */ function isSecretKey(input: string): boolean { return classifyKeyImportInput(input.toLowerCase()) !== "unknown"; diff --git a/desktop/src/features/admin-console/AdminConsoleReportsTab.tsx b/desktop/src/features/admin-console/AdminConsoleReportsTab.tsx index 2c5cb7157c..e8276cc4c8 100644 --- a/desktop/src/features/admin-console/AdminConsoleReportsTab.tsx +++ b/desktop/src/features/admin-console/AdminConsoleReportsTab.tsx @@ -407,7 +407,7 @@ function ResolveReportForm({ const payload = frozenRef.current; try { - const resolution = await resolveAdminReport(origin, report.id, { + const resolution = await resolveAdminReport(origin, report, { action: payload.action, requestId: payload.requestId, expirationSecs: payload.expirationSecs, diff --git a/desktop/src/features/admin-console/api.ts b/desktop/src/features/admin-console/api.ts index 684b77ba12..c07e31569f 100644 --- a/desktop/src/features/admin-console/api.ts +++ b/desktop/src/features/admin-console/api.ts @@ -9,7 +9,8 @@ * Callers must cancel in-flight queries on pubkey or origin change. */ -import { invokeTauri } from "@/shared/api/tauri"; +import { getRelayWsUrl, invokeTauri } from "@/shared/api/tauri"; +import { beginChannelMembershipWrite } from "@/shared/api/channelMembershipWrites"; import { invoke as invokeTauriRaw } from "@tauri-apps/api/core"; // ── Probe ───────────────────────────────────────────────────────────────── @@ -361,14 +362,27 @@ export type AdminReportResolution = { */ export async function resolveAdminReport( origin: string, - id: string, + report: Pick, body: AdminResolveReportBody, ): Promise { - return invokeTauri("admin_resolve_report", { - origin, - id, - body, - }); + const record = beginChannelMembershipWrite(); + const resolution = await invokeTauri( + "admin_resolve_report", + { origin, id: report.id, body }, + ); + // A kick removes the target from the report's channel. Reports are + // deployment-wide, so only a channel in the active community is recorded. + const action = resolution.activeAction; + if ( + action?.action === "kick" && + action.status === "succeeded" && + report.channelId && + communityHostFromRelayUrl(await getRelayWsUrl()) === + normalizeCommunityHost(report.communityHost) + ) { + record(report.channelId); + } + return resolution; } /** @@ -732,3 +746,27 @@ export async function directAdminAction( intent, }); } + +/** + * Normalize a community host the way the relay's `normalize_host` does: + * lowercase, no default port, no trailing root dot. + */ +export function normalizeCommunityHost(raw: string): string { + return raw + .trim() + .toLowerCase() + .replace(/:(443|80)$/, "") + .replace(/\.$/, ""); +} + +/** + * The community a relay URL serves is its `Host`: the URL's authority, with + * the default port already dropped for `ws`/`wss`. Null when unparseable. + */ +export function communityHostFromRelayUrl(relayUrl: string): string | null { + try { + return normalizeCommunityHost(new URL(relayUrl).host) || null; + } catch { + return null; + } +} diff --git a/desktop/src/features/agents/channelAgents.ts b/desktop/src/features/agents/channelAgents.ts index 24ace21b52..2437e921d5 100644 --- a/desktop/src/features/agents/channelAgents.ts +++ b/desktop/src/features/agents/channelAgents.ts @@ -460,7 +460,9 @@ export async function createChannelManagedAgents( ); const [managedAgents, members, personas] = await Promise.all([ listManagedAgents(), - getChannelMembers(channelId), + // Read-your-writes: templates call this right after creating the channel, + // before a read replica may have the member list. + getChannelMembers(channelId, { readYourWrites: true }), needsPersonaPolicy ? listPersonas() : Promise.resolve([]), ]); const channelMemberPubkeys = new Set( diff --git a/desktop/src/features/agents/hooks.ts b/desktop/src/features/agents/hooks.ts index 421525a4ff..1b479a7630 100644 --- a/desktop/src/features/agents/hooks.ts +++ b/desktop/src/features/agents/hooks.ts @@ -35,12 +35,12 @@ import { getManagedAgentLog, getRuntimeFileConfig, installAcpRuntime, - invokeTauri, listManagedAgents, listRelayAgents, saveCustomHarness, updateManagedAgent, } from "@/shared/api/tauri"; +import { syncAgentsToActiveHuddle } from "@/shared/api/tauriChannels"; import { discoverAcpCommands } from "@/shared/api/acpCommands"; import type { HarnessDefinitionInput } from "@/shared/api/tauri"; import { discoverAcpRuntimes } from "@/shared/api/tauriAcpDiscovery"; @@ -726,10 +726,9 @@ export function useAttachManagedAgentToChannelMutation( pubkey: result.agent.pubkey, }), ); - void invokeTauri("sync_agents_to_active_huddle", { - channelId: effectiveChannelId, - agentPubkeys: [result.agent.pubkey], - }).catch((error) => { + void syncAgentsToActiveHuddle(effectiveChannelId, [ + result.agent.pubkey, + ]).catch((error) => { console.warn("Could not sync attached agent into Huddle:", error); }); }, diff --git a/desktop/src/features/channels/hooks.ts b/desktop/src/features/channels/hooks.ts index fb43d00cd9..aca9f35b92 100644 --- a/desktop/src/features/channels/hooks.ts +++ b/desktop/src/features/channels/hooks.ts @@ -18,7 +18,6 @@ import { joinChannel, leaveChannel, openDm, - invokeTauri, removeChannelMember, setChannelPurpose, setChannelTopic, @@ -34,9 +33,10 @@ import type { SetChannelTopicInput, UpdateChannelInput, } from "@/shared/api/types"; -import type { - GetChannelsPayload, - OpenDmInput, +import { + syncAgentsToActiveHuddle, + type GetChannelsPayload, + type OpenDmInput, } from "@/shared/api/tauriChannels"; import { mergeConcurrentChannelRecency } from "@/features/channels/lib/channelRecencyMerge"; import { useIdentityQuery } from "@/shared/api/hooks"; @@ -845,12 +845,11 @@ export function useAddChannelMembersMutation(channelId: string | null) { variables.role === "bot" && result.added.length > 0 ) { - void invokeTauri("sync_agents_to_active_huddle", { - channelId: effectiveChannelId, - agentPubkeys: result.added, - }).catch((error) => { - console.warn("Could not sync added agents into Huddle:", error); - }); + void syncAgentsToActiveHuddle(effectiveChannelId, result.added).catch( + (error) => { + console.warn("Could not sync added agents into Huddle:", error); + }, + ); } }, onSettled: async (_data, _err, variables) => { diff --git a/desktop/src/features/channels/rosterAdminKick.test.mjs b/desktop/src/features/channels/rosterAdminKick.test.mjs new file mode 100644 index 0000000000..919a835668 --- /dev/null +++ b/desktop/src/features/channels/rosterAdminKick.test.mjs @@ -0,0 +1,202 @@ +// Admin Console kicks bypass the channel-event wrappers, so the member list +// must still refresh from the writer: from the relay's live `admin_kick` +// system message, and from the initiating resolve when it confirms the kick. +import assert from "node:assert/strict"; +import { after, mock, test } from "node:test"; +import { + QueryClient, + QueryClientProvider, + QueryObserver, +} from "@tanstack/react-query"; +import { JSDOM } from "jsdom"; +import React from "react"; + +const dom = new JSDOM("", { + url: "http://localhost", +}); +Object.assign(globalThis, { + document: dom.window.document, + HTMLElement: dom.window.HTMLElement, + IS_REACT_ACT_ENVIRONMENT: true, + window: dom.window, +}); +after(() => dom.window.close()); + +const OWNER = "aa".repeat(32); +const BOT = "bb".repeat(32); +const id = "00000000-0000-0000-0000-0000000000c1"; +let writer; +let replica; +let routes; +// Commands whose reply is held until the test releases it. +const held = new Map(); +window.__TAURI_INTERNALS__ = { + invoke: async (command, args) => { + await held.get(command)?.promise; + if (command === "get_relay_ws_url") return "wss://alpha.example.com"; + if (command === "get_channel_window") return []; + if (command === "get_channel_members") { + const strong = args.readYourWrites === true; + routes.push(strong ? "writer" : "replica"); + return { + members: (strong ? writer : replica).map((pubkey) => ({ + pubkey, + role: "member", + })), + }; + } + if (command === "add_channel_members") { + writer = [OWNER]; + return { added: [], errors: [] }; + } + if (command === "admin_resolve_report") { + writer = [OWNER]; + return { + status: "resolved", + activeAction: { action: "kick", status: "succeeded" }, + }; + } + return null; + }, +}; + +const { useChannelSubscription } = await import("@/features/messages/hooks"); +const { relayClient } = await import("@/shared/api/relayClient"); +const { addChannelMembers, getChannelMembers } = await import( + "@/shared/api/tauri" +); +const { resolveAdminReport } = await import("@/features/admin-console/api"); +const { resetChannelMembershipWrites, shouldReadChannelMembersFromWriter } = + await import("@/shared/api/channelMembershipWrites"); +const { channelMembersQueryKey, refreshRostersOnMembershipChange } = + await import("./rosterFreshness.ts"); +const { KIND_SYSTEM_MESSAGE } = await import("@/shared/constants/kinds"); + +async function settle() { + for (let i = 0; i < 10; i++) await new Promise((r) => setTimeout(r, 0)); +} + +// Mounts a cached owner+bot roster, then runs `act` and returns the roster. +async function rosterAfter(act) { + resetChannelMembershipWrites(); + writer = [OWNER, BOT]; + replica = [OWNER, BOT]; + routes = []; + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const stop = refreshRostersOnMembershipChange(client); + const observer = new QueryObserver(client, { + queryKey: channelMembersQueryKey(id), + queryFn: () => getChannelMembers(id), + staleTime: 300_000, + }); + const unsubscribe = observer.subscribe(() => {}); + try { + await settle(); + await act(client); + await settle(); + return client + .getQueryData(channelMembersQueryKey(id)) + ?.map((member) => member.pubkey); + } finally { + unsubscribe(); + stop(); + client.clear(); + mock.restoreAll(); + } +} + +test("a live admin_kick system message refreshes the roster from the writer", async () => { + const { renderHook, act, cleanup } = await import("@testing-library/react"); + let receive; + mock.method(relayClient, "subscribeToReconnects", () => () => {}); + mock.method(relayClient, "subscribeToChannelLive", async (_, cb) => { + receive = cb; + return async () => {}; + }); + const channel = { + id, + name: "general", + channelType: "stream", + visibility: "open", + isMember: true, + }; + const shown = await rosterAfter(async (client) => { + const view = renderHook(() => useChannelSubscription(channel), { + wrapper: ({ children }) => + React.createElement(QueryClientProvider, { client }, children), + }); + await act(async () => await settle()); + writer = [OWNER]; + await act(async () => { + receive({ + id: "kick-event", + pubkey: OWNER, + created_at: 20, + kind: KIND_SYSTEM_MESSAGE, + tags: [["h", id]], + content: JSON.stringify({ type: "admin_kick", target: BOT }), + sig: "", + }); + await settle(); + }); + view.unmount(); + cleanup(); + }); + assert.deepEqual(shown, [OWNER]); + assert.equal(routes.at(-1), "writer"); +}); + +for (const { host, recorded } of [ + { host: "Alpha.example.com:443", recorded: true }, + { host: "beta.example.com", recorded: false }, +]) { + test(`a confirmed report kick in ${host} ${recorded ? "refreshes" : "leaves"} the active community's roster`, async () => { + const shown = await rosterAfter(() => + resolveAdminReport( + "https://admin.invalid", + { id: "report", channelId: id, communityHost: host }, + { action: "kick", requestId: "request" }, + ), + ); + assert.deepEqual(shown, recorded ? [OWNER] : [OWNER, BOT]); + assert.deepEqual(routes, recorded ? ["replica", "writer"] : ["replica"]); + }); +} + +const kick = () => + resolveAdminReport( + "https://admin.invalid", + { id: "report", channelId: id, communityHost: "alpha.example.com" }, + { action: "kick", requestId: "request" }, + ); +const addMember = () => + addChannelMembers({ channelId: id, pubkeys: [BOT], role: "member" }); + +// A write started before a community switch must not mark its channel in the +// new community, whichever reply is still in flight when the switch happens. +for (const { name, write, pending } of [ + { name: "a report kick", write: kick, pending: "admin_resolve_report" }, + { name: "a report kick", write: kick, pending: "get_relay_ws_url" }, + { name: "an added member", write: addMember, pending: "add_channel_members" }, +]) { + test(`${name} whose ${pending} reply lands after a community reset records nothing`, async () => { + const shown = await rosterAfter(async () => { + let release; + held.set(pending, { promise: new Promise((r) => (release = r)) }); + try { + const write$ = write(); + await settle(); + resetChannelMembershipWrites(); + release(); + await write$; + } finally { + held.delete(pending); + } + }); + assert.equal(shouldReadChannelMembersFromWriter(id), false); + assert.deepEqual(routes, ["replica"]); + assert.deepEqual(shown, [OWNER, BOT]); + }); +} diff --git a/desktop/src/features/channels/rosterFreshness.test.mjs b/desktop/src/features/channels/rosterFreshness.test.mjs new file mode 100644 index 0000000000..17815bf8ef --- /dev/null +++ b/desktop/src/features/channels/rosterFreshness.test.mjs @@ -0,0 +1,344 @@ +// After this client changes a channel's membership, the member list shown +// must never settle on the list from before the change, even though the +// relay's replica keeps returning that stale list for a while. +import assert from "node:assert/strict"; +import test from "node:test"; + +import { QueryClient, QueryObserver } from "@tanstack/react-query"; + +let now = 1_000_000; +// The window runs on the monotonic clock; wall-clock jumps must not move it. +Object.defineProperty(performance, "now", { value: () => now }); + +const OWNER = "aa".repeat(32); +const BOT = "bb".repeat(32); +const routes = []; +let writerMembers = [OWNER]; +let replicaMembers = [OWNER]; +let nativeResults = {}; +const deferred = []; +let deferNextRead = false; + +globalThis.window = { + setTimeout, + clearTimeout, + __TAURI_INTERNALS__: { + invoke: async (command, args) => { + if (command === "get_channel_members") { + const writer = args.readYourWrites === true; + routes.push(writer ? "writer" : "replica"); + const members = writer ? writerMembers : replicaMembers; + const response = { + members: members.map((pubkey) => ({ pubkey, role: "member" })), + }; + if (deferNextRead) { + deferNextRead = false; + return new Promise((resolve) => + deferred.push(() => resolve(response)), + ); + } + return response; + } + if (command === "add_channel_members") { + writerMembers = [...writerMembers, ...args.pubkeys]; + return { added: args.pubkeys, errors: [] }; + } + if (command in nativeResults) { + writerMembers = [...writerMembers, BOT]; + return nativeResults[command]; + } + if (command === "join_channel") { + writerMembers = [...writerMembers, BOT]; + return undefined; + } + return undefined; + }, + }, +}; + +const { addChannelMembers, joinChannel, getChannelMembers } = await import( + "@/shared/api/tauri" +); +const { ensureStarterChannels, syncAgentsToActiveHuddle } = await import( + "@/shared/api/tauriChannels" +); +const { resolveBestieConversation } = await import( + "@/protectedFeatures/bestie/api" +); +const { noteChannelMembershipChange, resetChannelMembershipWrites } = + await import("@/shared/api/channelMembershipWrites"); +const { channelsQueryKey, invalidateChannelState } = await import("./hooks.ts"); +const { channelMembersQueryKey, refreshRostersOnMembershipChange } = + await import("./rosterFreshness.ts"); + +function setup(channelId) { + routes.length = 0; + writerMembers = [OWNER]; + replicaMembers = [OWNER]; + nativeResults = {}; + now += 60_000; + const queryClient = new QueryClient(); + const unsubscribeRefresh = refreshRostersOnMembershipChange(queryClient); + // Same query function as useChannelMembersQuery. + const observer = new QueryObserver(queryClient, { + queryKey: channelMembersQueryKey(channelId), + queryFn: () => getChannelMembers(channelId), + staleTime: 5 * 60_000, + }); + return { + queryClient, + observer, + mount: () => observer.subscribe(() => {}), + shown: () => + queryClient + .getQueryData(channelMembersQueryKey(channelId)) + ?.map((member) => member.pubkey), + teardown: unsubscribeRefresh, + }; +} + +async function settle() { + for (let i = 0; i < 10; i += 1) await new Promise((r) => setTimeout(r, 0)); +} + +test("adding a member then invalidating channel state keeps the new roster", async () => { + const roster = setup("ch-add"); + const unmount = roster.mount(); + await settle(); + assert.deepEqual(roster.shown(), [OWNER]); + + await addChannelMembers({ channelId: "ch-add", pubkeys: [BOT], role: "bot" }); + await invalidateChannelState(roster.queryClient, "ch-add"); + await settle(); + + assert.deepEqual(roster.shown(), [OWNER, BOT]); + assert.deepEqual( + routes.slice(1), + routes.slice(1).map(() => "writer"), + ); + assert.ok(routes.length > 1); + unmount(); + roster.teardown(); +}); + +test("joining from the browser then opening the channel shows the joined roster", async () => { + const roster = setup("ch-join"); + // AppShell's browser join handler, then the dialog opens the channel. + await joinChannel("ch-join"); + await roster.queryClient.invalidateQueries({ queryKey: channelsQueryKey }); + const unmount = roster.mount(); + await settle(); + + assert.deepEqual(roster.shown(), [OWNER, BOT]); + assert.deepEqual(routes, ["writer"]); + unmount(); + roster.teardown(); +}); + +test("a replica read in flight before the change cannot land over it", async () => { + const roster = setup("ch-race"); + deferNextRead = true; + const unmount = roster.mount(); + await settle(); + assert.deepEqual(routes, ["replica"]); + + await addChannelMembers({ + channelId: "ch-race", + pubkeys: [BOT], + role: "bot", + }); + await settle(); + for (const resolve of deferred.splice(0)) resolve(); + await settle(); + + assert.deepEqual(roster.shown(), [OWNER, BOT]); + unmount(); + roster.teardown(); +}); + +test("the window outlasts the relay's 30s staleness ceiling", async () => { + const roster = setup("ch-ceiling"); + const unmount = roster.mount(); + await settle(); + await addChannelMembers({ + channelId: "ch-ceiling", + pubkeys: [BOT], + role: "bot", + }); + await settle(); + // A 30s replica budget may still legally serve the old list here. + now += 30_000; + await roster.queryClient.invalidateQueries({ + queryKey: channelMembersQueryKey("ch-ceiling"), + }); + await settle(); + assert.deepEqual(roster.shown(), [OWNER, BOT]); + assert.equal(routes.at(-1), "writer"); + + // Past the ceiling the replica has caught up, and reads return to it. + now += 1_001; + replicaMembers = [OWNER, BOT]; + await roster.queryClient.invalidateQueries({ + queryKey: channelMembersQueryKey("ch-ceiling"), + }); + await settle(); + assert.deepEqual(roster.shown(), [OWNER, BOT]); + assert.equal(routes.at(-1), "replica"); + unmount(); + roster.teardown(); +}); + +test("a wall-clock jump does not end the window early", async () => { + const roster = setup("ch-clock"); + await addChannelMembers({ + channelId: "ch-clock", + pubkeys: [BOT], + role: "bot", + }); + const realDateNow = Date.now; + Date.now = () => realDateNow() + 3_600_000; + try { + await getChannelMembers("ch-clock"); + } finally { + Date.now = realDateNow; + } + assert.equal(routes.at(-1), "writer"); + roster.teardown(); +}); + +async function cacheOldRoster(roster) { + const unmount = roster.mount(); + await settle(); + assert.deepEqual(roster.shown(), [OWNER]); + return unmount; +} + +test("starter joins that landed are recorded when a later join fails", async () => { + const roster = setup("starter-general"); + const unmount = await cacheOldRoster(roster); + nativeResults.ensure_starter_channels = { + channels: [], + changed_channel_ids: ["starter-general"], + error: "starter join rejected", + }; + await assert.rejects(ensureStarterChannels(), { + message: "starter join rejected", + }); + await settle(); + assert.deepEqual(roster.shown(), [OWNER, BOT]); + unmount(); + roster.teardown(); +}); + +test("huddle sync records the parent channel it changed", async () => { + const roster = setup("huddle-parent"); + const unmount = await cacheOldRoster(roster); + nativeResults.sync_agents_to_active_huddle = { + matched_active_huddle: true, + added: [BOT], + changed_channel_ids: ["huddle-ephemeral", "huddle-parent"], + error: null, + }; + await syncAgentsToActiveHuddle("huddle-ephemeral", [BOT]); + await settle(); + assert.deepEqual(roster.shown(), [OWNER, BOT]); + unmount(); + roster.teardown(); +}); + +test("huddle sync records changes made before it failed", async () => { + const roster = setup("huddle-parent-partial"); + const unmount = await cacheOldRoster(roster); + nativeResults.sync_agents_to_active_huddle = { + matched_active_huddle: true, + added: [], + changed_channel_ids: ["huddle-parent-partial"], + error: "agent add rejected", + }; + await assert.rejects( + syncAgentsToActiveHuddle("huddle-parent-partial", [BOT]), + { + message: "agent add rejected", + }, + ); + await settle(); + assert.deepEqual(roster.shown(), [OWNER, BOT]); + unmount(); + roster.teardown(); +}); + +test("a Bestie DM's first roster read goes to the writer", async () => { + const roster = setup("bestie-dm"); + nativeResults.resolve_bestie_conversation = { + id: "bestie-dm", + name: "bestie", + channel_type: "dm", + visibility: "private", + participants: [], + participant_pubkeys: [], + }; + await resolveBestieConversation({ + expectedRelayUrl: "wss://relay.test", + expectedSignerPubkey: OWNER, + }); + const unmount = roster.mount(); + await settle(); + assert.deepEqual(routes, ["writer"]); + assert.deepEqual(roster.shown(), [OWNER, BOT]); + unmount(); + roster.teardown(); +}); + +test("a change recorded before the listener registers still refreshes", async () => { + routes.length = 0; + const queryClient = new QueryClient(); + // Cached old roster, then a change while no listener exists. + queryClient.setQueryData(channelMembersQueryKey("early"), [ + { pubkey: OWNER, role: "member" }, + ]); + const observer = new QueryObserver(queryClient, { + queryKey: channelMembersQueryKey("early"), + queryFn: () => getChannelMembers("early"), + staleTime: 5 * 60_000, + }); + const unmount = observer.subscribe(() => {}); + writerMembers = [OWNER, BOT]; + noteChannelMembershipChange("early"); + const teardown = refreshRostersOnMembershipChange(queryClient); + await settle(); + assert.deepEqual( + queryClient + .getQueryData(channelMembersQueryKey("early")) + ?.map((member) => member.pubkey), + [OWNER, BOT], + ); + unmount(); + teardown(); +}); + +test("ordinary reads return to the replica once the window passes", async () => { + const roster = setup("ch-later"); + await addChannelMembers({ + channelId: "ch-later", + pubkeys: [BOT], + role: "bot", + }); + now += 31_001; + await getChannelMembers("ch-later"); + await getChannelMembers("ch-other"); + assert.deepEqual(routes.slice(-2), ["replica", "replica"]); + roster.teardown(); +}); + +test("a community switch clears recorded changes", async () => { + const roster = setup("ch-community"); + await addChannelMembers({ + channelId: "ch-community", + pubkeys: [BOT], + role: "bot", + }); + resetChannelMembershipWrites(); + await getChannelMembers("ch-community"); + assert.equal(routes.at(-1), "replica"); + roster.teardown(); +}); diff --git a/desktop/src/features/channels/rosterFreshness.ts b/desktop/src/features/channels/rosterFreshness.ts index a0a3603165..3d23091263 100644 --- a/desktop/src/features/channels/rosterFreshness.ts +++ b/desktop/src/features/channels/rosterFreshness.ts @@ -6,6 +6,8 @@ import type { useQueryClient } from "@tanstack/react-query"; +import { onChannelMembershipChange } from "@/shared/api/channelMembershipWrites"; + /** Single source for the members cache key; hooks.ts imports it from here. */ export const channelMembersQueryKey = (channelId: string) => ["channels", channelId, "members"] as const; @@ -20,7 +22,8 @@ export const channelMembersQueryKey = (channelId: string) => * a third party joining a channel the viewer is not currently subscribed to, * which corrects within this window. The previous 30s window put a full * roster fetch (kind:39002 + a kind:0 batch over every member) on nearly - * every channel switch. + * every channel switch. Refetches just after a change read from the relay + * writer (channelMembershipWrites), so none can cache the pre-change roster. */ export const CHANNEL_MEMBERS_STALE_TIME_MS = 5 * 60_000; @@ -43,3 +46,22 @@ export async function invalidateChannelMembersRosters( }); } } + +/** + * Refreshes a channel's cached roster whenever its membership changes (see + * channelMembershipWrites). Cancels a fetch already in flight first: it may + * have been routed to the lagging replica before the change was recorded. + */ +export function refreshRostersOnMembershipChange( + queryClient: Pick< + ReturnType, + "cancelQueries" | "invalidateQueries" + >, +) { + return onChannelMembershipChange((channelId) => { + const queryKey = channelMembersQueryKey(channelId); + void queryClient + .cancelQueries({ queryKey, exact: true }) + .then(() => queryClient.invalidateQueries({ queryKey, exact: true })); + }); +} diff --git a/desktop/src/features/channels/useMembershipNotifications.ts b/desktop/src/features/channels/useMembershipNotifications.ts index 7f8fa91a6a..8302b37d42 100644 --- a/desktop/src/features/channels/useMembershipNotifications.ts +++ b/desktop/src/features/channels/useMembershipNotifications.ts @@ -2,6 +2,7 @@ import * as React from "react"; import { useQueryClient } from "@tanstack/react-query"; import { channelsQueryKey } from "@/features/channels/hooks"; +import { noteChannelMembershipChange } from "@/shared/api/channelMembershipWrites"; import { getChannelIdFromTags } from "@/features/messages/lib/threading"; import { relayClient } from "@/shared/api/relayClient"; import type { RelayEvent } from "@/shared/api/types"; @@ -29,9 +30,7 @@ export function useMembershipNotifications(currentPubkey?: string) { void queryClient.invalidateQueries({ queryKey: ["channels", channelId, "detail"], }); - void queryClient.invalidateQueries({ - queryKey: ["channels", channelId, "members"], - }); + noteChannelMembershipChange(channelId); }, ); diff --git a/desktop/src/features/communities/useCommunityInit.ts b/desktop/src/features/communities/useCommunityInit.ts index 665dfb96ed..a9f0b1f1fc 100644 --- a/desktop/src/features/communities/useCommunityInit.ts +++ b/desktop/src/features/communities/useCommunityInit.ts @@ -3,6 +3,7 @@ import { isTauri } from "@tauri-apps/api/core"; import { isMacPlatform } from "@/shared/lib/platform"; import { relayClient } from "@/shared/api/relayClient"; +import { resetChannelMembershipWrites } from "@/shared/api/channelMembershipWrites"; import { resetRateLimitGate } from "@/shared/api/relayRateLimitGate"; import { autoConnectDefaultRelayEnabled, @@ -88,6 +89,7 @@ async function resetCommunityState({ resetBackgroundMediaUploads(); resetLinkPreviewPreparations(); resetPersistentAgentAudienceStore(); + resetChannelMembershipWrites(); // Intentionally NOT reset: the in-flight detached agent-start map // (`useDetachedAgentStart`). Its entries are keyed by the scope each start // asserts (relay URL + signer + agent pubkey), so they cannot leak into the diff --git a/desktop/src/features/huddle/components/HuddleBar.tsx b/desktop/src/features/huddle/components/HuddleBar.tsx index 4c01ca68a1..57dfa1fada 100644 --- a/desktop/src/features/huddle/components/HuddleBar.tsx +++ b/desktop/src/features/huddle/components/HuddleBar.tsx @@ -33,6 +33,7 @@ import { AddAgentDialog, type AgentAddResult } from "./AddAgentDialog"; import type { HuddleAgentVoiceSettings } from "./AgentVoiceMenu"; import { MicControls, SpeakerControls } from "./MicControls"; import { HuddleParticipantsControl } from "./ParticipantList"; +import { beginChannelMembershipWrite } from "@/shared/api/channelMembershipWrites"; import { truncateNpub } from "@/shared/lib/pubkey"; // Mirrors HuddleState in src-tauri/src/huddle/mod.rs. @@ -640,11 +641,19 @@ export function HuddleBar({ onClose={() => setShowAddAgent(false)} onAdd={async (pubkey: string): Promise => { setAgentAddError(null); + const record = beginChannelMembershipWrite(); try { const result = await invoke( "add_agent_to_huddle", { agentPubkey: pubkey }, ); + if (barState?.ephemeral_channel_id) { + record(barState.ephemeral_channel_id); + } + // The agent may also have been added to the parent channel. + if (result.parent_added && barState?.parent_channel_id) { + record(barState.parent_channel_id); + } // Refresh huddle state so the participant list updates immediately. const s = await invoke("get_huddle_state"); setState(s); @@ -719,10 +728,14 @@ export function HuddleBar({ "Remove this agent from the huddle?", ); if (!confirmed) return; + const record = beginChannelMembershipWrite(); try { await invoke("remove_agent_from_huddle", { agentPubkey: pubkey, }); + if (barState?.ephemeral_channel_id) { + record(barState.ephemeral_channel_id); + } setState((prev) => { if (!prev) return prev; return { diff --git a/desktop/src/features/huddle/components/HuddleRoomHeader.tsx b/desktop/src/features/huddle/components/HuddleRoomHeader.tsx index 0cf9c735da..13a6a371a8 100644 --- a/desktop/src/features/huddle/components/HuddleRoomHeader.tsx +++ b/desktop/src/features/huddle/components/HuddleRoomHeader.tsx @@ -3,6 +3,7 @@ import { listen } from "@tauri-apps/api/event"; import * as React from "react"; import { useProfileQuery, useSelfProfileCache } from "@/features/profile/hooks"; +import { beginChannelMembershipWrite } from "@/shared/api/channelMembershipWrites"; import { useIdentityQuery } from "@/shared/api/hooks"; import { useHuddle, useHuddleLevels } from "../HuddleContext"; import { useHuddleParticipantRoster } from "../hooks/useHuddleParticipantRoster"; @@ -53,29 +54,36 @@ export function HuddleRoomHeader() { } return levels; }, [currentPubkey, isMuted, micConnected, micLevel, speakerLevels]); - const handleRemoveAgent = React.useCallback(async (pubkey: string) => { - if (!window.confirm("Remove this agent from the huddle?")) return; - try { - await invoke("remove_agent_from_huddle", { - agentPubkey: pubkey, - }); - setState((current) => - current - ? { - ...current, - participants: current.participants.filter( - (member) => member !== pubkey, - ), - agent_pubkeys: current.agent_pubkeys.filter( - (agent) => agent !== pubkey, - ), - } - : current, - ); - } catch (error) { - console.error("Failed to remove agent from huddle:", error); - } - }, []); + const handleRemoveAgent = React.useCallback( + async (pubkey: string) => { + if (!window.confirm("Remove this agent from the huddle?")) return; + const record = beginChannelMembershipWrite(); + try { + await invoke("remove_agent_from_huddle", { + agentPubkey: pubkey, + }); + if (state?.ephemeral_channel_id) { + record(state.ephemeral_channel_id); + } + setState((current) => + current + ? { + ...current, + participants: current.participants.filter( + (member) => member !== pubkey, + ), + agent_pubkeys: current.agent_pubkeys.filter( + (agent) => agent !== pubkey, + ), + } + : current, + ); + } catch (error) { + console.error("Failed to remove agent from huddle:", error); + } + }, + [state?.ephemeral_channel_id], + ); React.useEffect(() => { let disposed = false; diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index d28f292608..cbd6a8eea4 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -59,6 +59,7 @@ import { getChannelWindowEvents } from "@/shared/api/channelWindow"; import type { Channel, Identity, RelayEvent } from "@/shared/api/types"; // Same .mjs the renderer uses, so the cache-update projection can't drift // from the on-render overlay. +import { noteChannelMembershipChange } from "@/shared/api/channelMembershipWrites"; import { applyEditTagOverlay } from "@/features/messages/lib/applyEditTagOverlay.mjs"; import { emptyChannelWindowStore, @@ -377,11 +378,10 @@ export function useChannelSubscription(channel: Channel | null) { if ( payload.type === "member_joined" || payload.type === "member_left" || - payload.type === "member_removed" + payload.type === "member_removed" || + payload.type === "admin_kick" ) { - void queryClient.invalidateQueries({ - queryKey: ["channels", channelId, "members"], - }); + noteChannelMembershipChange(channelId); void queryClient.invalidateQueries({ queryKey: ["channels"], exact: true, diff --git a/desktop/src/features/messages/ui/useMentionSendFlow.test-support.mjs b/desktop/src/features/messages/ui/useMentionSendFlow.test-support.mjs index bb5fc3de53..893f9fd523 100644 --- a/desktop/src/features/messages/ui/useMentionSendFlow.test-support.mjs +++ b/desktop/src/features/messages/ui/useMentionSendFlow.test-support.mjs @@ -156,7 +156,7 @@ export async function setup({ lifecycle = false } = {}) { "@/features/messages/lib/imetaMediaMarkdown": { buildOutgoingMessage: (text) => ({ content: text, mediaTags: [] }), }, - "@/shared/api/tauri": { invokeTauri: async () => {} }, + "@/shared/api/tauriChannels": { syncAgentsToActiveHuddle: async () => {} }, "@/shared/lib/pubkey": { normalizePubkey: (key) => key.toLowerCase(), truncatePubkey: (key) => key, diff --git a/desktop/src/features/messages/ui/useMentionSendFlow.ts b/desktop/src/features/messages/ui/useMentionSendFlow.ts index 299c0144ab..a74645d122 100644 --- a/desktop/src/features/messages/ui/useMentionSendFlow.ts +++ b/desktop/src/features/messages/ui/useMentionSendFlow.ts @@ -26,7 +26,7 @@ import { import { useActivePreparedLinkPreviews } from "./useActivePreparedLinkPreviews"; import { useDetachedAgentStart } from "./useDetachedAgentStart"; import { useEnsureAgentMentionsReady } from "./useEnsureAgentMentionsReady"; -import { invokeTauri } from "@/shared/api/tauri"; +import { syncAgentsToActiveHuddle } from "@/shared/api/tauriChannels"; import type { AcpRuntime, ManagedAgent } from "@/shared/api/types"; import { normalizePubkey, truncateNpub } from "@/shared/lib/pubkey"; import { buildCustomEmojiTags } from "@/shared/lib/customEmojiTags"; @@ -510,10 +510,7 @@ export function useMentionSendFlow({ } if (preparedAgentPubkeys.length > 0 && sendChannelId) { try { - await invokeTauri("sync_agents_to_active_huddle", { - channelId: sendChannelId, - agentPubkeys: preparedAgentPubkeys, - }); + await syncAgentsToActiveHuddle(sendChannelId, preparedAgentPubkeys); if (isSendCancelled()) return restoreComposerAfterFailure(); } catch (error) { if (isSendCancelled()) return restoreComposerAfterFailure(); diff --git a/desktop/src/protectedFeatures/bestie/api.ts b/desktop/src/protectedFeatures/bestie/api.ts index b29b046fdf..bac9357f00 100644 --- a/desktop/src/protectedFeatures/bestie/api.ts +++ b/desktop/src/protectedFeatures/bestie/api.ts @@ -1,3 +1,4 @@ +import { beginChannelMembershipWrite } from "@/shared/api/channelMembershipWrites"; import type { Channel } from "@/shared/api/types"; import { fromRawChannel, type RawChannel } from "@/shared/api/tauriChannels"; import { invokeTauri } from "@/shared/api/tauri"; @@ -50,7 +51,11 @@ export async function clearBestieAssignment(scope: BestieScope): Promise { export async function resolveBestieConversation( scope: BestieScope, ): Promise { - return fromRawChannel( + const record = beginChannelMembershipWrite(); + const channel = fromRawChannel( await invokeTauri("resolve_bestie_conversation", scope), ); + // May open a new DM, as openDm does. + record(channel.id); + return channel; } diff --git a/desktop/src/shared/api/channelMembershipWrites.ts b/desktop/src/shared/api/channelMembershipWrites.ts new file mode 100644 index 0000000000..1cf0ce50e7 --- /dev/null +++ b/desktop/src/shared/api/channelMembershipWrites.ts @@ -0,0 +1,77 @@ +/** + * Read-your-writes routing for channel member lists. The relay serves + * ordinary reads from a replica that can lag the writer, so a member-list + * read made just after a membership change can return the list from before + * it, and the roster cache keeps that result fresh for minutes. For a short + * window after this client changes a channel's membership (or observes a + * live membership change), every member-list read of that channel is routed + * to the writer, so no refresh in that window, however it is triggered, can + * land a pre-change list. + */ + +/** + * Just over the relay's hard staleness ceiling: a replica may serve a read + * only while its last verified sync is within the configured budget, and the + * budget is clamped to `FENCE_STALENESS` (30s, in + * crates/buzz-db/src/runtime/replica_fence.rs). Past this, no supported + * setting can return a pre-change list. Must move with `FENCE_STALENESS`. + * Measured on the monotonic clock so a wall-clock jump can't shorten it. + */ +const WRITER_READ_WINDOW_MS = 31_000; + +const writerReadDeadlines = new Map(); +let generation = 0; +const listeners = new Set<(channelId: string) => void>(); + +/** + * Records a membership change for `channelId`: its member-list reads go to + * the writer for the window, and listeners (the app's roster cache) refresh. + */ +export function noteChannelMembershipChange(channelId: string) { + const now = performance.now(); + for (const [id, deadline] of writerReadDeadlines) { + if (deadline <= now) writerReadDeadlines.delete(id); + } + writerReadDeadlines.set(channelId, now + WRITER_READ_WINDOW_MS); + for (const listener of listeners) listener(channelId); +} + +/** + * Starts a membership write. Call before the write's first await; the + * returned recorder notes a changed channel only if no community reset has + * happened since, so a write that settles after a switch can't mark a + * channel in the new community. + */ +export function beginChannelMembershipWrite(): (channelId: string) => void { + const started = generation; + return (channelId) => { + if (started === generation) noteChannelMembershipChange(channelId); + }; +} + +export function shouldReadChannelMembersFromWriter(channelId: string) { + return (writerReadDeadlines.get(channelId) ?? 0) > performance.now(); +} + +/** + * Subscribes to membership changes, first replaying every channel whose + * window is still open so a change recorded before registration still + * refreshes a roster cached before it. + */ +export function onChannelMembershipChange( + listener: (channelId: string) => void, +): () => void { + for (const channelId of writerReadDeadlines.keys()) { + if (shouldReadChannelMembersFromWriter(channelId)) listener(channelId); + } + listeners.add(listener); + return () => { + listeners.delete(listener); + }; +} + +/** Community reset: changes recorded on the old relay don't apply to the new one. */ +export function resetChannelMembershipWrites() { + generation++; + writerReadDeadlines.clear(); +} diff --git a/desktop/src/shared/api/tauri.ts b/desktop/src/shared/api/tauri.ts index 34101dc7c5..093f9e62ed 100644 --- a/desktop/src/shared/api/tauri.ts +++ b/desktop/src/shared/api/tauri.ts @@ -1,4 +1,5 @@ import { invoke as tauriInvoke } from "@tauri-apps/api/core"; +import { beginChannelMembershipWrite } from "@/shared/api/channelMembershipWrites"; import { fromRawInstallRuntimeResult, type RawInstallRuntimeResult, @@ -243,7 +244,7 @@ export class TauriInvokeError extends Error { } } -function toTauriError(error: unknown): Error { +export function toTauriError(error: unknown): Error { if (error instanceof Error) { return error; } @@ -348,14 +349,22 @@ export function getRelayHttpUrl(): Promise { export async function addChannelMembers( input: AddChannelMembersInput, ): Promise { - return invokeTauri("add_channel_members", input); + const record = beginChannelMembershipWrite(); + const result = await invokeTauri( + "add_channel_members", + input, + ); + record(input.channelId); + return result; } export async function removeChannelMember( channelId: string, pubkey: string, ): Promise { + const record = beginChannelMembershipWrite(); await invokeTauri("remove_channel_member", { channelId, pubkey }); + record(channelId); } export async function changeChannelMemberRole( @@ -363,15 +372,21 @@ export async function changeChannelMemberRole( pubkey: string, role: string, ): Promise { + const record = beginChannelMembershipWrite(); await invokeTauri("change_channel_member_role", { channelId, pubkey, role }); + record(channelId); } export async function joinChannel(channelId: string): Promise { + const record = beginChannelMembershipWrite(); await invokeTauri("join_channel", { channelId }); + record(channelId); } export async function leaveChannel(channelId: string): Promise { + const record = beginChannelMembershipWrite(); await invokeTauri("leave_channel", { channelId }); + record(channelId); } export async function getHomeFeed( diff --git a/desktop/src/shared/api/tauriChannels.ts b/desktop/src/shared/api/tauriChannels.ts index ca7a71c2ce..a497cb839f 100644 --- a/desktop/src/shared/api/tauriChannels.ts +++ b/desktop/src/shared/api/tauriChannels.ts @@ -10,7 +10,11 @@ import type { SetChannelTopicInput, UpdateChannelInput, } from "@/shared/api/types"; -import { invokeTauri } from "@/shared/api/tauri"; +import { + beginChannelMembershipWrite, + shouldReadChannelMembersFromWriter, +} from "@/shared/api/channelMembershipWrites"; +import { invokeTauri, toTauriError } from "@/shared/api/tauri"; export type RawChannel = { id: string; @@ -162,13 +166,43 @@ export async function getOpenChannelDirectory(): Promise { export async function createChannel( input: CreateChannelInput, ): Promise { - return fromRawChannel(await invokeTauri("create_channel", input)); + const record = beginChannelMembershipWrite(); + const channel = fromRawChannel( + await invokeTauri("create_channel", input), + ); + record(channel.id); + return channel; } export async function ensureStarterChannels(): Promise { - return (await invokeTauri("ensure_starter_channels")).map( - fromRawChannel, - ); + const record = beginChannelMembershipWrite(); + // Channels created or joined are reported even when a later step fails. + const result = await invokeTauri<{ + channels: RawChannel[]; + changed_channel_ids: string[]; + error: string | null; + }>("ensure_starter_channels"); + for (const id of result.changed_channel_ids) record(id); + if (result.error !== null) throw toTauriError(result.error); + return result.channels.map(fromRawChannel); +} + +/** + * Enrolls channel agents into the active Huddle. Native sync can add them to + * the Huddle's ephemeral and parent channels, whichever `channelId` names, so + * it reports every channel it changed, including before a failure. + */ +export async function syncAgentsToActiveHuddle( + channelId: string, + agentPubkeys: string[], +): Promise { + const record = beginChannelMembershipWrite(); + const result = await invokeTauri<{ + changed_channel_ids: string[]; + error: string | null; + }>("sync_agents_to_active_huddle", { channelId, agentPubkeys }); + for (const id of result.changed_channel_ids) record(id); + if (result.error !== null) throw toTauriError(result.error); } export type OpenDmInput = { @@ -191,7 +225,12 @@ export type OpenDmInput = { }; export async function openDm(input: OpenDmInput): Promise { - return fromRawChannel(await invokeTauri("open_dm", input)); + const record = beginChannelMembershipWrite(); + const channel = fromRawChannel( + await invokeTauri("open_dm", input), + ); + record(channel.id); + return channel; } export async function hideDm(channelId: string): Promise { @@ -285,14 +324,23 @@ export async function getChannelMessagesBefore( export async function getChannelMembers( channelId: string, + options?: { readYourWrites?: boolean }, ): Promise { const response = await invokeTauri( "get_channel_members", - { channelId }, + { + channelId, + readYourWrites: + options?.readYourWrites || + shouldReadChannelMembersFromWriter(channelId) || + undefined, + }, ); return response.members.map(fromRawChannelMember); } export async function joinChannel(channelId: string): Promise { + const record = beginChannelMembershipWrite(); await invokeTauri("join_channel", { channelId }); + record(channelId); } diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 201701952a..3e0328f4ab 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -7221,7 +7221,11 @@ async function handleGetPresence( async function handleEnsureStarterChannels( config: E2eConfig | undefined, -): Promise { +): Promise<{ + channels: RawChannelWithMembership[]; + changed_channel_ids: string[]; + error: null; +}> { const starterChannelError = config?.mock?.ensureStarterChannelsErrors?.shift(); if (starterChannelError) { @@ -7229,6 +7233,7 @@ async function handleEnsureStarterChannels( } const currentPubkey = getMockMemberPubkey(config); + const changedChannelIds: string[] = []; const ensureMember = (channel: MockChannel) => { if ( channel.members.some( @@ -7242,6 +7247,7 @@ async function handleEnsureStarterChannels( channel.members.push(createCurrentMember(config, "member")); syncMockChannel(channel); touchMockChannel(channel); + changedChannelIds.push(channel.id); }; for (const channelName of [ @@ -7261,7 +7267,11 @@ async function handleEnsureStarterChannels( ensureMember(channel); } - return listMockChannels(config); + return { + channels: listMockChannels(config), + changed_channel_ids: changedChannelIds, + error: null, + }; } async function handleCreateChannel( @@ -7332,7 +7342,7 @@ async function handleCreateChannel( // Fetch the created channel via pure Nostr query. // The relay emits kind:39000 as a side effect of kind:9007. const metaEvents = await relayQuery(config, [ - { kinds: [39000], "#d": [channelId], limit: 1 }, + { kinds: [39000], "#d": [channelId], limit: 1, consistency: "strong" }, ]); const ev = metaEvents[0]; if (!ev) { @@ -7440,7 +7450,7 @@ async function handleOpenDm( // Fetch channel metadata const metaEvents = await relayQuery(config, [ - { kinds: [39000], "#d": [channelId], limit: 1 }, + { kinds: [39000], "#d": [channelId], limit: 1, consistency: "strong" }, ]); const ev = metaEvents[0]; const evTags = (ev?.tags ?? []) as string[][]; @@ -7536,7 +7546,7 @@ async function handleGetChannelDetails( } async function handleGetChannelMembers( - args: { channelId: string }, + args: { channelId: string; readYourWrites?: boolean }, config: E2eConfig | undefined, ): Promise { const delayMs = config?.mock?.channelMembersReadDelayMs ?? 0; @@ -7554,7 +7564,12 @@ async function handleGetChannelMembers( } const memberEvents = await relayQuery(config, [ - { kinds: [39002], "#d": [args.channelId], limit: 1 }, + { + kinds: [39002], + "#d": [args.channelId], + limit: 1, + ...(args.readYourWrites ? { consistency: "strong" } : {}), + }, ]); const memberTags = ((memberEvents[0]?.tags ?? []) as string[][]).filter( (t) => t[0] === "p", @@ -7630,7 +7645,7 @@ async function handleUpdateChannel( // Re-fetch updated metadata const metaEvents = await relayQuery(config, [ - { kinds: [39000], "#d": [args.channelId], limit: 1 }, + { kinds: [39000], "#d": [args.channelId], limit: 1, consistency: "strong" }, ]); const ev = metaEvents[0]; const evTags = (ev?.tags ?? []) as string[][]; @@ -12252,7 +12267,12 @@ export function maybeInstallE2eTauriMocks() { (request.channelId !== mockHuddle.state.parent_channel_id && request.channelId !== mockHuddle.state.ephemeral_channel_id) ) { - return { matched_active_huddle: false, added: [] }; + return { + matched_active_huddle: false, + added: [], + changed_channel_ids: [], + error: null, + }; } const existingAgents = new Set( @@ -12272,7 +12292,13 @@ export function maybeInstallE2eTauriMocks() { if (added.length > 0) { await emitMockHuddleState(); } - return { matched_active_huddle: true, added }; + return { + matched_active_huddle: true, + added, + changed_channel_ids: + added.length > 0 ? [mockHuddle.state.ephemeral_channel_id] : [], + error: null, + }; } case "add_agent_to_huddle": { const error = activeConfig?.mock?.addAgentToHuddleError; diff --git a/desktop/tests/e2e/huddle-transcription.spec.ts b/desktop/tests/e2e/huddle-transcription.spec.ts index 254fd5ef5b..e6c58bed8d 100644 --- a/desktop/tests/e2e/huddle-transcription.spec.ts +++ b/desktop/tests/e2e/huddle-transcription.spec.ts @@ -926,6 +926,8 @@ test("adds channel-mentioned agents to the live huddle roster", async ({ expect(result).toEqual({ matched_active_huddle: true, added: [TEST_IDENTITIES.bob.pubkey], + changed_channel_ids: [HUDDLE_CHANNEL_ID], + error: null, }); await expect(participantTiles).toHaveCount(3); await expect(