From 215f451306fff73c4a04c1274bafcc2bac52b9a4 Mon Sep 17 00:00:00 2001 From: Orual Date: Sat, 6 Dec 2025 15:03:58 -0500 Subject: [PATCH] cursor presence and real-time doc sync --- Cargo.lock | 1 - crates/weaver-app/Cargo.toml | 2 +- crates/weaver-app/assets/styling/editor.css | 6 + crates/weaver-app/src/collab_context.rs | 31 +- .../src/components/editor/component.rs | 100 ++++-- .../src/components/editor/cursor.rs | 120 +++++++ .../weaver-app/src/components/editor/sync.rs | 302 ++++++++++++++++-- crates/weaver-app/src/main.rs | 3 +- crates/weaver-common/Cargo.toml | 1 - crates/weaver-common/src/agent.rs | 120 +++++++ .../weaver-common/src/transport/messages.rs | 77 ++++- crates/weaver-common/src/transport/mod.rs | 5 +- crates/weaver-common/src/transport/node.rs | 56 +++- crates/weaver-common/src/transport/session.rs | 123 +++++-- 14 files changed, 852 insertions(+), 95 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fcfe0a9..02e2b78 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11563,7 +11563,6 @@ dependencies = [ "http", "iroh", "iroh-gossip", - "iroh-quinn", "iroh-tickets", "jacquard", "jacquard-common", diff --git a/crates/weaver-app/Cargo.toml b/crates/weaver-app/Cargo.toml index 2b6128c..592a116 100644 --- a/crates/weaver-app/Cargo.toml +++ b/crates/weaver-app/Cargo.toml @@ -80,7 +80,7 @@ mini-moka = { git = "https://github.com/moka-rs/mini-moka", rev = "da864e849f5d0 chrono = { version = "0.4", features = ["wasmbind"] } wasm-bindgen = "0.2" wasm-bindgen-futures = "0.4" -web-sys = { version = "0.3", features = ["ServiceWorkerContainer", "ServiceWorker", "ServiceWorkerRegistration", "RegistrationOptions", "Window", "Navigator", "MessageEvent", "console", "Document", "Element", "HtmlImageElement", "Selection", "Range", "Node", "HtmlElement", "TreeWalker", "NodeFilter", "DomTokenList", "Clipboard", "ClipboardItem", "Blob", "BlobPropertyBag", "EventTarget", "InputEvent", "AddEventListenerOptions"] } +web-sys = { version = "0.3", features = ["ServiceWorkerContainer", "ServiceWorker", "ServiceWorkerRegistration", "RegistrationOptions", "Window", "Navigator", "MessageEvent", "console", "Document", "Element", "HtmlImageElement", "Selection", "Range", "Node", "HtmlElement", "TreeWalker", "NodeFilter", "DomTokenList", "Clipboard", "ClipboardItem", "Blob", "BlobPropertyBag", "EventTarget", "InputEvent", "AddEventListenerOptions", "DomRect", "DomRectList"] } js-sys = "0.3" gloo-storage = "0.3" gloo-timers = "0.3" diff --git a/crates/weaver-app/assets/styling/editor.css b/crates/weaver-app/assets/styling/editor.css index 7165df5..91a27b7 100644 --- a/crates/weaver-app/assets/styling/editor.css +++ b/crates/weaver-app/assets/styling/editor.css @@ -1188,6 +1188,12 @@ li:has(.md-syntax-block:not(.hidden)) { pointer-events: none; } +.remote-selection { + position: absolute; + pointer-events: none; + border-radius: 2px; +} + .remote-cursor-caret { width: 2px; height: var(--cursor-height, 18px); diff --git a/crates/weaver-app/src/collab_context.rs b/crates/weaver-app/src/collab_context.rs index 595d0c5..4eda0d3 100644 --- a/crates/weaver-app/src/collab_context.rs +++ b/crates/weaver-app/src/collab_context.rs @@ -9,6 +9,25 @@ use dioxus::prelude::*; use std::sync::Arc; use weaver_common::transport::CollabNode; +/// Debug state for the collab session, displayed in editor debug panel. +#[derive(Clone, Default)] +pub struct CollabDebugState { + /// Our node ID + pub node_id: Option, + /// Our relay URL + pub relay_url: Option, + /// URI of our published session record + pub session_record_uri: Option, + /// Number of discovered peers + pub discovered_peers: usize, + /// Number of connected peers + pub connected_peers: usize, + /// Whether we've joined the gossip swarm + pub is_joined: bool, + /// Last error message + pub last_error: Option, +} + /// Context state for the collaboration node. /// /// This is provided as a Dioxus context and can be accessed by editor components @@ -39,6 +58,7 @@ impl Default for CollabContext { #[component] pub fn CollabProvider(children: Element) -> Element { let mut collab_ctx = use_signal(CollabContext::default); + let debug_state = use_signal(CollabDebugState::default); // Spawn the CollabNode on mount let _spawn_result = use_resource(move || async move { @@ -62,8 +82,9 @@ pub fn CollabProvider(children: Element) -> Element { } }); - // Provide the context + // Provide the contexts use_context_provider(|| collab_ctx); + use_context_provider(|| debug_state); rsx! { {children} } } @@ -74,7 +95,9 @@ pub fn CollabProvider(children: Element) -> Element { pub fn CollabProvider(children: Element) -> Element { // On server/native, provide an empty context (collab happens in browser) let collab_ctx = use_signal(CollabContext::default); + let debug_state = use_signal(CollabDebugState::default); use_context_provider(|| collab_ctx); + use_context_provider(|| debug_state); rsx! { {children} } } @@ -91,3 +114,9 @@ pub fn use_collab_available() -> bool { let ctx = use_context::>(); ctx.read().node.is_some() } + +/// Hook to get the collab debug state signal. +/// Returns None if called outside CollabProvider. +pub fn try_use_collab_debug() -> Option> { + try_use_context::>() +} diff --git a/crates/weaver-app/src/components/editor/component.rs b/crates/weaver-app/src/components/editor/component.rs index 4b403f2..9f73ac2 100644 --- a/crates/weaver-app/src/components/editor/component.rs +++ b/crates/weaver-app/src/components/editor/component.rs @@ -561,13 +561,11 @@ fn MarkdownEditorInner( let mut doc_for_dom = document.clone(); #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] use_effect(move || { - tracing::debug!("DOM update effect triggered"); - - tracing::debug!( - composition_active = doc_for_dom.composition.read().is_some(), - cursor = doc_for_dom.cursor.read().offset, - "DOM update: checking state" - ); + // tracing::debug!( + // composition_active = doc_for_dom.composition.read().is_some(), + // cursor = doc_for_dom.cursor.read().offset, + // "DOM update: checking state" + // ); // Skip DOM updates during IME composition - browser controls the preview if doc_for_dom.composition.read().is_some() { @@ -575,7 +573,7 @@ fn MarkdownEditorInner( return; } - tracing::debug!( + tracing::trace!( cursor = doc_for_dom.cursor.read().offset, len = doc_for_dom.len_chars(), "DOM update proceeding (not in composition)" @@ -1087,7 +1085,7 @@ fn MarkdownEditorInner( onselect: { let mut doc = document.clone(); move |_evt| { - tracing::debug!("onselect fired"); + tracing::trace!("onselect fired"); let paras = cached_paragraphs(); sync_cursor_from_dom(&mut doc, editor_id, ¶s); let spans = syntax_spans(); @@ -1105,7 +1103,7 @@ fn MarkdownEditorInner( onselectstart: { let mut doc = document.clone(); move |_evt| { - tracing::debug!("onselectstart fired"); + tracing::trace!("onselectstart fired"); let paras = cached_paragraphs(); sync_cursor_from_dom(&mut doc, editor_id, ¶s); let spans = syntax_spans(); @@ -1123,7 +1121,7 @@ fn MarkdownEditorInner( onselectionchange: { let mut doc = document.clone(); move |_evt| { - tracing::debug!("onselectionchange fired"); + tracing::trace!("onselectionchange fired"); let paras = cached_paragraphs(); sync_cursor_from_dom(&mut doc, editor_id, ¶s); let spans = syntax_spans(); @@ -1141,7 +1139,7 @@ fn MarkdownEditorInner( onclick: { let mut doc = document.clone(); move |evt| { - tracing::debug!("onclick fired"); + tracing::trace!("onclick fired"); let paras = cached_paragraphs(); // Check if click target is a math-clickable element @@ -1255,7 +1253,7 @@ fn MarkdownEditorInner( let mut doc = document.clone(); move |evt: CompositionEvent| { let data = evt.data().data(); - tracing::debug!( + tracing::trace!( data = %data, "compositionstart" ); @@ -1264,7 +1262,7 @@ fn MarkdownEditorInner( if let Some(sel) = sel { let (start, end) = (sel.anchor.min(sel.head), sel.anchor.max(sel.head)); - tracing::debug!( + tracing::trace!( start, end, "compositionstart: deleting selection" @@ -1274,7 +1272,7 @@ fn MarkdownEditorInner( } let cursor_offset = doc.cursor.read().offset; - tracing::debug!( + tracing::trace!( cursor = cursor_offset, "compositionstart: setting composition state" ); @@ -1289,7 +1287,7 @@ fn MarkdownEditorInner( let mut doc = document.clone(); move |evt: CompositionEvent| { let data = evt.data().data(); - tracing::debug!( + tracing::trace!( data = %data, "compositionupdate" ); @@ -1306,7 +1304,7 @@ fn MarkdownEditorInner( let mut doc = document.clone(); move |evt: CompositionEvent| { let final_text = evt.data().data(); - tracing::debug!( + tracing::trace!( data = %final_text, "compositionend" ); @@ -1354,6 +1352,28 @@ fn MarkdownEditorInner( } div { class: "editor-debug", div { "Cursor: {document.cursor.read().offset}, Chars: {document.len_chars()}" }, + // Collab debug info + { + if let Some(debug_state) = crate::collab_context::try_use_collab_debug() { + let ds = debug_state.read(); + rsx! { + div { class: "collab-debug", + if let Some(ref node_id) = ds.node_id { + span { title: "{node_id}", "Node: {&node_id[..8.min(node_id.len())]}…" } + } + if ds.is_joined { + span { class: "joined", "✓ Joined" } + } + span { "Peers: {ds.discovered_peers}" } + if let Some(ref err) = ds.last_error { + span { class: "error", title: "{err}", "⚠" } + } + } + } + } else { + rsx! {} + } + }, ReportButton { email: "editor-bugs@weaver.sh".to_string(), editor_id: "markdown-editor".to_string(), @@ -1516,11 +1536,20 @@ fn RemoteCursors( render_cache: Signal, ) -> Element { let presence_read = presence.read(); + let cursor_count = presence_read.len(); let cursors: Vec<_> = presence_read .cursors() .map(|(c, cur)| (c.display_name.clone(), c.color, cur.position, cur.selection)) .collect(); + if cursor_count > 0 { + tracing::debug!( + "RemoteCursors: {} collaborators, {} with cursors", + cursor_count, + cursors.len() + ); + } + if cursors.is_empty() { return rsx! {}; } @@ -1558,28 +1587,61 @@ fn RemoteCursorIndicator( color: u32, offset_map: Vec, ) -> Element { - use super::cursor::{get_cursor_rect_relative, CursorRect}; + use super::cursor::{get_cursor_rect_relative, get_selection_rects_relative}; - // Convert RGBA u32 to CSS color + // Convert RGBA u32 to CSS color (fully opaque for cursor) let r = (color >> 24) & 0xFF; let g = (color >> 16) & 0xFF; let b = (color >> 8) & 0xFF; let a = (color & 0xFF) as f32 / 255.0; let color_css = format!("rgba({}, {}, {}, {})", r, g, b, a); + // Semi-transparent version for selection highlight + let selection_color_css = format!("rgba({}, {}, {}, 0.25)", r, g, b); // Get cursor position relative to editor let rect = get_cursor_rect_relative(position, &offset_map, "markdown-editor"); + // Get selection rectangles if there's a selection + let selection_rects = if let Some((start, end)) = selection { + let (start, end) = if start <= end { (start, end) } else { (end, start) }; + get_selection_rects_relative(start, end, &offset_map, "markdown-editor") + } else { + vec![] + }; + let Some(rect) = rect else { + tracing::debug!( + "RemoteCursorIndicator: no rect for position {} (offset_map len: {})", + position, + offset_map.len() + ); return rsx! {}; }; + tracing::trace!( + "RemoteCursorIndicator: {} at ({}, {}) h={}, selection_rects={}", + display_name, + rect.x, + rect.y, + rect.height, + selection_rects.len() + ); + let style = format!( "left: {}px; top: {}px; --cursor-height: {}px; --cursor-color: {};", rect.x, rect.y, rect.height, color_css ); rsx! { + // Selection highlight rectangles (rendered behind cursor) + for (i, sel_rect) in selection_rects.iter().enumerate() { + div { + key: "sel-{i}", + class: "remote-selection", + style: "left: {sel_rect.x}px; top: {sel_rect.y}px; width: {sel_rect.width}px; height: {sel_rect.height}px; background-color: {selection_color_css};", + } + } + div { class: "remote-cursor", style: "{style}", diff --git a/crates/weaver-app/src/components/editor/cursor.rs b/crates/weaver-app/src/components/editor/cursor.rs index 7bd1b36..2ed0a41 100644 --- a/crates/weaver-app/src/components/editor/cursor.rs +++ b/crates/weaver-app/src/components/editor/cursor.rs @@ -307,3 +307,123 @@ pub fn get_cursor_rect_relative( ) -> Option { None } + +/// A rectangle for part of a selection (one per line). +#[derive(Debug, Clone, Copy)] +pub struct SelectionRect { + pub x: f64, + pub y: f64, + pub width: f64, + pub height: f64, +} + +/// Get screen rectangles for a selection range, relative to editor. +/// +/// Returns multiple rects if selection spans multiple lines. +#[cfg(all(target_family = "wasm", target_os = "unknown"))] +pub fn get_selection_rects_relative( + start: usize, + end: usize, + offset_map: &[OffsetMapping], + editor_id: &str, +) -> Vec { + use wasm_bindgen::JsCast; + + if offset_map.is_empty() || start >= end { + return vec![]; + } + + let Some(window) = web_sys::window() else { + return vec![]; + }; + let Some(document) = window.document() else { + return vec![]; + }; + let Some(editor) = document.get_element_by_id(editor_id) else { + return vec![]; + }; + let editor_rect = editor.get_bounding_client_rect(); + + // Find mappings for start and end + let Some((start_mapping, _)) = find_mapping_for_char(offset_map, start) else { + return vec![]; + }; + let Some((end_mapping, _)) = find_mapping_for_char(offset_map, end) else { + return vec![]; + }; + + // Get containers + let start_container = document + .get_element_by_id(&start_mapping.node_id) + .or_else(|| { + let selector = format!("[data-node-id='{}']", start_mapping.node_id); + document.query_selector(&selector).ok().flatten() + }); + let end_container = document + .get_element_by_id(&end_mapping.node_id) + .or_else(|| { + let selector = format!("[data-node-id='{}']", end_mapping.node_id); + document.query_selector(&selector).ok().flatten() + }); + + let (Some(start_container), Some(end_container)) = (start_container, end_container) else { + return vec![]; + }; + + // Create range + let Ok(range) = document.create_range() else { + return vec![]; + }; + + // Set start + if let Some(child_index) = start_mapping.child_index { + let _ = range.set_start(&start_container, child_index as u32); + } else if let Ok(container_element) = start_container.clone().dyn_into::() { + let offset_in_range = start - start_mapping.char_range.start; + let target_utf16_offset = start_mapping.char_offset_in_node + offset_in_range; + if let Ok((text_node, node_offset)) = find_text_node_at_offset(&container_element, target_utf16_offset) { + let _ = range.set_start(&text_node, node_offset as u32); + } + } + + // Set end + if let Some(child_index) = end_mapping.child_index { + let _ = range.set_end(&end_container, child_index as u32); + } else if let Ok(container_element) = end_container.dyn_into::() { + let offset_in_range = end - end_mapping.char_range.start; + let target_utf16_offset = end_mapping.char_offset_in_node + offset_in_range; + if let Ok((text_node, node_offset)) = find_text_node_at_offset(&container_element, target_utf16_offset) { + let _ = range.set_end(&text_node, node_offset as u32); + } + } + + // Get all rects (one per line) + let Some(rects) = range.get_client_rects() else { + return vec![]; + }; + let mut result = Vec::new(); + + for i in 0..rects.length() { + if let Some(rect) = rects.get(i) { + let rect: web_sys::DomRect = rect; + result.push(SelectionRect { + x: rect.x() - editor_rect.x(), + y: rect.y() - editor_rect.y(), + width: rect.width(), + height: rect.height().max(16.0), + }); + } + } + + result +} + +#[cfg(not(all(target_family = "wasm", target_os = "unknown")))] +pub fn get_selection_rects_relative( + _start: usize, + _end: usize, + _offset_map: &[OffsetMapping], + _editor_id: &str, +) -> Vec { + vec![] +} diff --git a/crates/weaver-app/src/components/editor/sync.rs b/crates/weaver-app/src/components/editor/sync.rs index d721e3a..f8c22df 100644 --- a/crates/weaver-app/src/components/editor/sync.rs +++ b/crates/weaver-app/src/components/editor/sync.rs @@ -103,8 +103,6 @@ async fn prefetch_embeds_from_doc( if let Ok(records_container) = embeds_map.get_or_create_container("records", loro::LoroList::new()) { - tracing::debug!("Loro embeds map records len: {}", records_container.len()); - for i in 0..records_container.len() { let Some(value) = records_container.get(i) else { continue; @@ -132,7 +130,6 @@ async fn prefetch_embeds_from_doc( .await { Ok(html) => { - tracing::debug!("Pre-fetched embed from Loro map: {}", key_uri); resolved.add_embed(key_uri, html, None); } Err(e) => { @@ -148,13 +145,12 @@ async fn prefetch_embeds_from_doc( // Strategy 2: If no embeds found in Loro map, parse markdown text if resolved.embed_content.is_empty() { - use weaver_common::{collect_refs_from_markdown, ExtractedRef}; + use weaver_common::{ExtractedRef, collect_refs_from_markdown}; let text = doc.get_text("content"); let markdown = text.to_string(); if !markdown.is_empty() { - tracing::debug!("Falling back to markdown parsing for embeds"); let refs = collect_refs_from_markdown(&markdown); for extracted in refs { @@ -166,7 +162,6 @@ async fn prefetch_embeds_from_doc( match weaver_renderer::atproto::fetch_and_render(&key_uri, fetcher).await { Ok(html) => { - tracing::debug!("Pre-fetched embed from markdown: {}", uri); resolved.add_embed(key_uri, html, None); } Err(e) => { @@ -707,10 +702,8 @@ pub async fn create_diff( // Use inline for small diffs, blob for larger ones let (blob_ref, inline_diff): (Option>, _) = if updates.len() <= INLINE_THRESHOLD { - tracing::debug!("Using inline diff ({} bytes)", updates.len()); (None, Some(jacquard::bytes::Bytes::from(updates))) } else { - tracing::debug!("Using blob diff ({} bytes)", updates.len()); let mime_type = MimeType::new_static("application/octet-stream"); let blob = client.upload_blob(updates, mime_type).await.map_err(|e| { WeaverError::InvalidNotebook(format!("Failed to upload diff: {}", e)) @@ -976,13 +969,15 @@ pub async fn load_all_edit_states_from_pds( // Get the last seen diff rkey for this root (if any) let after_rkey = root_uri.as_ref().and_then(|uri| { - last_seen_diffs.get(uri).and_then(|diff_uri| { - diff_uri.rkey().map(|rk| rk.0.to_string()) - }) + last_seen_diffs + .get(uri) + .and_then(|diff_uri| diff_uri.rkey().map(|rk| rk.0.to_string())) }); // Load state from this root (skipping already-seen diffs) - if let Some(pds_state) = load_edit_state_from_root_id(fetcher, root_id, after_rkey.as_deref()).await? { + if let Some(pds_state) = + load_edit_state_from_root_id(fetcher, root_id, after_rkey.as_deref()).await? + { // Import root snapshot into merged doc if let Err(e) = merged_doc.import(&pds_state.root_snapshot) { tracing::warn!("Failed to import root snapshot from {}: {:?}", root_did, e); @@ -1112,7 +1107,7 @@ async fn load_edit_state_from_root_id( // Skip diffs we've already seen (rkey/TID is lexicographically sortable by time) if let Some(after) = after_rkey { if rkey_str <= after { - tracing::debug!("Skipping already-seen diff rkey: {}", rkey_str); + tracing::trace!("Skipping already-seen diff rkey: {}", rkey_str); continue; } } @@ -1365,15 +1360,18 @@ const SESSION_TTL_MINUTES: u32 = 15; #[cfg(all(target_arch = "wasm32", target_os = "unknown"))] #[component] pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { + use crate::collab_context::try_use_collab_debug; use tokio::sync::mpsc; - use weaver_common::transport::{CollabMessage, CollabSession, SessionEvent}; use weaver_common::WeaverExt; + use weaver_common::transport::{CollabMessage, CollabSession, SessionEvent}; let collab_node = use_collab_node(); let fetcher = use_context::(); let mut session: Signal>> = use_signal(|| None); // URI of our published session record (for cleanup) let mut session_record_uri: Signal>> = use_signal(|| None); + // Debug state for display in editor debug panel (optional, may not be provided) + let debug_state = try_use_collab_debug(); // Channel for sending local updates from Loro callback to async broadcast task let mut update_tx: Signal>>> = use_signal(|| None); // Channel for sending cursor updates @@ -1438,13 +1436,15 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { cursor_tx.set(Some(ctx)); // Subscribe to local updates from Loro - fires when local changes are committed - let sub = doc_for_join.loro_doc().subscribe_local_update(Box::new(move |update| { - tracing::debug!("RealTimeSync: local update ({} bytes)", update.len()); - if let Err(e) = tx.send(update.to_vec()) { - tracing::warn!("RealTimeSync: failed to queue update: {}", e); - } - true // Keep subscription active - })); + let sub = doc_for_join + .loro_doc() + .subscribe_local_update(Box::new(move |update| { + tracing::debug!("RealTimeSync: local update ({} bytes)", update.len()); + if let Err(e) = tx.send(update.to_vec()) { + tracing::warn!("RealTimeSync: failed to queue update: {}", e); + } + true // Keep subscription active + })); _subscription.set(Some(sub)); let doc_for_recv = doc_for_join.clone(); @@ -1455,10 +1455,38 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { // Derive topic from resource URI let topic = CollabSession::topic_from_uri(uri.as_str()); + // Wait for relay connection before discovering peers or publishing session + // Browser clients REQUIRE relay for peer connectivity + let relay_url = node.wait_for_relay().await; + tracing::info!( + relay_url = %relay_url, + "RealTimeSync: relay connection ready" + ); + + // Update debug state with node info + if let Some(mut ds) = debug_state { + ds.with_mut(|s| { + s.node_id = Some(node.node_id_string()); + s.relay_url = Some(relay_url.clone()); + }); + } + // Discover existing session peers for bootstrap let bootstrap_peers = match fetcher.find_session_peers(&uri).await { Ok(peers) => { tracing::info!("RealTimeSync: found {} existing peers", peers.len()); + if let Some(mut ds) = debug_state { + ds.with_mut(|s| s.discovered_peers = peers.len()); + } + for p in &peers { + tracing::info!( + did = %p.did, + node_id = %p.node_id, + relay_url = ?p.relay_url, + expires_at = ?p.expires_at, + "RealTimeSync: discovered peer" + ); + } peers .into_iter() .filter_map(|p| { @@ -1468,6 +1496,9 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { } Err(e) => { tracing::warn!("RealTimeSync: failed to find peers: {}", e); + if let Some(mut ds) = debug_state { + ds.with_mut(|s| s.last_error = Some(format!("peer discovery: {}", e))); + } vec![] } }; @@ -1478,27 +1509,111 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { .create_collab_session( &resource_ref_for_spawn, &node_id_str, - None, // relay_url - could add if needed + Some(&relay_url), Some(SESSION_TTL_MINUTES), ) .await { Ok(uri) => { tracing::info!("RealTimeSync: published session record: {}", uri); + if let Some(mut ds) = debug_state { + ds.with_mut(|s| s.session_record_uri = Some(uri.to_string())); + } session_record_uri.set(Some(uri)); } Err(e) => { tracing::warn!("RealTimeSync: failed to publish session record: {}", e); + if let Some(mut ds) = debug_state { + ds.with_mut(|s| s.last_error = Some(format!("publish session: {}", e))); + } } } + // Clone before join() consumes them + let node_for_discovery = node.clone(); + let bootstrap_peers_set = bootstrap_peers.clone(); + match CollabSession::join(node, topic, bootstrap_peers).await { Ok((collab_session, mut event_stream)) => { let collab_session = Arc::new(collab_session); session.set(Some(collab_session.clone())); + if let Some(mut ds) = debug_state { + ds.with_mut(|s| s.is_joined = true); + } tracing::info!("RealTimeSync: joined session for {}", uri); + // Broadcast Join message to announce ourselves + let our_did = fetcher.current_did().await; + let display_name = if let Some(ref did) = our_did { + use jacquard::types::ident::AtIdentifier; + use weaver_api::sh_weaver::actor::ProfileDataViewInner; + + let ident = AtIdentifier::Did(did.clone()); + fetcher + .fetch_profile(&ident) + .await + .ok() + .and_then(|p| match &p.inner { + ProfileDataViewInner::ProfileView(pv) => { + pv.display_name.as_ref().map(|s| s.to_string()) + } + ProfileDataViewInner::ProfileViewDetailed(pv) => { + pv.display_name.as_ref().map(|s| s.to_string()) + } + _ => None, + }) + .unwrap_or_else(|| "Collaborator".to_string()) + } else { + "Collaborator".to_string() + }; + let join_msg = CollabMessage::Join { + did: our_did.map(|d| d.to_string()).unwrap_or_default(), + display_name, + }; + if let Err(e) = collab_session.broadcast(&join_msg).await { + tracing::warn!("RealTimeSync: failed to broadcast Join: {}", e); + } + + // Request sync from existing peers + // Convert our version vector to wire format + let our_vv = doc_for_recv.version_vector(); + let have_version: Vec<(u64, u64)> = our_vv + .iter() + .map(|(peer, counter)| (*peer, *counter as u64)) + .collect(); + let sync_request = CollabMessage::SyncRequest { have_version }; + if let Err(e) = collab_session.broadcast(&sync_request).await { + tracing::warn!("RealTimeSync: failed to broadcast SyncRequest: {}", e); + } else { + tracing::debug!("RealTimeSync: sent sync request ({} vv entries)", our_vv.len()); + } + + // Spawn TTL refresh task - keeps our session record alive + let session_uri_for_refresh = session_record_uri.clone(); + let fetcher_for_refresh = fetcher.clone(); + spawn(async move { + // Refresh every 5 minutes (TTL is 15 min, so plenty of buffer) + let mut interval = n0_future::time::interval( + n0_future::time::Duration::from_secs(5 * 60), + ); + loop { + interval.tick().await; + if let Some(uri) = session_uri_for_refresh.peek().clone() { + tracing::debug!("RealTimeSync: refreshing session TTL"); + if let Err(e) = fetcher_for_refresh + .refresh_collab_session(&uri, SESSION_TTL_MINUTES) + .await + { + tracing::warn!( + "RealTimeSync: failed to refresh session: {}", + e + ); + } + } + } + }); + // Spawn broadcast task - sends local updates to gossip let session_for_broadcast = collab_session.clone(); spawn(async move { @@ -1532,12 +1647,92 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { } }); + // Spawn periodic peer discovery task + // This handles the race condition where peers publish sessions + // at different times and might miss each other on initial discovery + let session_for_discovery = collab_session.clone(); + let fetcher_for_discovery = fetcher.clone(); + let uri_for_discovery = uri.clone(); + let our_node_id = node_for_discovery.node_id(); + let mut known_peers: std::collections::HashSet = + bootstrap_peers_set.iter().cloned().collect(); + spawn(async move { + // Check for new peers every 30 seconds + let mut interval = + n0_future::time::interval(n0_future::time::Duration::from_secs(30)); + loop { + interval.tick().await; + tracing::debug!("RealTimeSync: periodic discovery tick"); + match fetcher_for_discovery + .find_session_peers(&uri_for_discovery) + .await + { + Ok(peers) => { + tracing::info!( + "RealTimeSync: periodic discovery found {} session records", + peers.len() + ); + for p in &peers { + tracing::debug!( + " - peer: {} (relay: {:?}, expires: {:?})", + p.node_id, + p.relay_url, + p.expires_at + ); + } + // Filter: parse node ID, exclude ourselves, exclude already known + let new_peers: Vec<_> = peers + .into_iter() + .filter_map(|p| { + weaver_common::transport::parse_node_id(&p.node_id) + .ok() + }) + .filter(|id| *id != our_node_id) + .filter(|id| !known_peers.contains(id)) + .collect(); + + if !new_peers.is_empty() { + tracing::info!( + "RealTimeSync: periodic discovery found {} NEW peers", + new_peers.len() + ); + for p in &new_peers { + known_peers.insert(*p); + } + if let Err(e) = + session_for_discovery.join_peers(new_peers).await + { + tracing::warn!( + "RealTimeSync: failed to join discovered peers: {}", + e + ); + } + } + } + Err(e) => { + tracing::warn!( + "RealTimeSync: periodic peer discovery failed: {}", + e + ); + } + } + } + }); + // Spawn event receiver task - receives updates from peers let mut doc_for_recv = doc_for_recv.clone(); + let session_for_sync = collab_session.clone(); spawn(async move { use n0_future::StreamExt; - while let Some(event) = event_stream.next().await { + while let Some(result) = event_stream.next().await { + let event = match result { + Ok(e) => e, + Err(e) => { + tracing::error!("RealTimeSync: event stream error: {}", e); + break; + } + }; match event { SessionEvent::Message { from, message } => { match message { @@ -1559,11 +1754,16 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { selection, .. } => { - presence.write().update_cursor( - &from, - position, - selection, - ); + // Add peer if not known (cursor might arrive before Join) + let mut p = presence.write(); + if !p.contains(&from) { + p.add_collaborator( + from, + "unknown".into(), + "Peer".into(), + ); + } + p.update_cursor(&from, position, selection); } CollabMessage::Join { did, display_name } => { tracing::info!( @@ -1586,9 +1786,51 @@ pub fn RealTimeSync(props: RealTimeSyncProps) -> Element { "RealTimeSync: sync request (have {} entries)", have_version.len() ); - // TODO: Send snapshot or updates based on their version + // Convert their version vector from wire format + let their_vv: loro::VersionVector = have_version + .into_iter() + .map(|(peer, counter)| (peer, counter as i32)) + .collect(); + + // Export updates they don't have + if let Some(data) = + doc_for_recv.export_updates_from(&their_vv) + { + tracing::info!( + "RealTimeSync: sending {} bytes to sync peer", + data.len() + ); + let response = CollabMessage::SyncResponse { + data, + is_snapshot: false, + }; + if let Err(e) = + session_for_sync.broadcast(&response).await + { + tracing::warn!( + "RealTimeSync: failed to send sync response: {}", + e + ); + } + } else { + tracing::debug!( + "RealTimeSync: no updates to send (peer is up to date)" + ); + } + } + CollabMessage::SyncResponse { data, is_snapshot } => { + tracing::info!( + "RealTimeSync: received sync response ({} bytes, snapshot: {})", + data.len(), + is_snapshot + ); + if let Err(e) = doc_for_recv.import_updates(&data) { + tracing::warn!( + "RealTimeSync: failed to import sync response: {:?}", + e + ); + } } - _ => {} } } SessionEvent::PeerJoined(peer) => { diff --git a/crates/weaver-app/src/main.rs b/crates/weaver-app/src/main.rs index 231420e..cbf78b1 100644 --- a/crates/weaver-app/src/main.rs +++ b/crates/weaver-app/src/main.rs @@ -148,7 +148,8 @@ fn main() { ); // Filter out noisy crates - let filter = EnvFilter::new("debug,loro_internal=warn"); + let filter = + EnvFilter::new("debug,loro_internal=warn,jacquard_identity=info,jacquard_common=info"); let reg = Registry::default() .with(filter) diff --git a/crates/weaver-common/Cargo.toml b/crates/weaver-common/Cargo.toml index b263a2b..d4c6edc 100644 --- a/crates/weaver-common/Cargo.toml +++ b/crates/weaver-common/Cargo.toml @@ -58,7 +58,6 @@ wasm-bindgen-futures = "0.4" send_wrapper = "0.6" wasmworker = "0.1" wasmworker-proc-macro = "0.1" -iroh-quinn = { version = "0.14", default-features = false } ring = { version = "0.17", default-features = false, features = ["wasm32_unknown_unknown_js"]} getrandom = { version = "0.3", default-features = false, features = ["wasm_js"] } diff --git a/crates/weaver-common/src/agent.rs b/crates/weaver-common/src/agent.rs index f07c1eb..c39fe02 100644 --- a/crates/weaver-common/src/agent.rs +++ b/crates/weaver-common/src/agent.rs @@ -1420,10 +1420,17 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { }; if !accept_output.records.is_empty() { + // Both parties in a valid invite+accept pair are authorized + let inviter_did = record_id.did.clone().into_static(); + collaborators.push(inviter_did); collaborators.push(invitee_did); } } + // Deduplicate (someone might appear in multiple pairs) + collaborators.sort(); + collaborators.dedup(); + Ok(collaborators) } } @@ -1906,6 +1913,9 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { use jacquard::types::string::Datetime; use weaver_api::sh_weaver::collab::session::Session; + // Clean up any expired sessions first + let _ = self.cleanup_expired_sessions().await; + let now_chrono = chrono::Utc::now().fixed_offset(); let now = Datetime::new(now_chrono); let expires_at = ttl_minutes.map(|mins| { @@ -1973,6 +1983,98 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { } } + /// Update the relay URL in an existing session record. + /// + /// Called when the relay connection changes during a session. + fn update_collab_session_relay<'a>( + &'a self, + session_uri: &'a AtUri<'a>, + relay_url: Option<&'a str>, + ) -> impl Future> + 'a { + async move { + use weaver_api::sh_weaver::collab::session::Session; + + let relay_uri = relay_url + .map(|url| jacquard::types::string::Uri::new(url)) + .transpose() + .map_err(|_| AgentError::from(ClientError::invalid_request("Invalid relay URL")))?; + + self.update_record::(session_uri, |session| { + session.relay_url = relay_uri.clone(); + }) + .await?; + Ok(()) + } + } + + /// Delete all expired session records for the current user. + /// + /// Called before creating a new session to clean up stale records. + fn cleanup_expired_sessions<'a>( + &'a self, + ) -> impl Future> + 'a + where + Self: Sized, + { + async move { + use jacquard::types::nsid::Nsid; + use weaver_api::com_atproto::repo::list_records::ListRecords; + use weaver_api::sh_weaver::collab::session::Session; + + let (did, _) = self.session_info().await.ok_or_else(|| { + AgentError::from(ClientError::invalid_request("No active session")) + })?; + let now = chrono::Utc::now(); + let mut deleted = 0u32; + + // List all our session records + let collection = + Nsid::new("sh.weaver.collab.session").map_err(WeaverError::AtprotoString)?; + let request = ListRecords::new() + .repo(did.clone()) + .collection(collection) + .limit(100) + .build(); + + let response = self.send(request).await.map_err(AgentError::from)?; + let output = response.into_output().map_err(|e| { + AgentError::from(ClientError::invalid_request(format!( + "Failed to list sessions: {}", + e + ))) + })?; + + for record in output.records { + if let Ok(session) = jacquard::from_data::(&record.value) { + // Check if expired + if let Some(ref expires_at) = session.expires_at { + let expires_str = expires_at.as_str(); + if let Ok(expires) = chrono::DateTime::parse_from_rfc3339(expires_str) { + if expires.with_timezone(&chrono::Utc) < now { + // Delete expired session + if let Some(rkey) = record.uri.rkey() { + if let Err(e) = + self.delete_record::(rkey.clone()).await + { + tracing::warn!("Failed to delete expired session: {}", e); + } else { + deleted += 1; + } + } + } + } + } + } + } + + if deleted > 0 { + tracing::info!("Cleaned up {} expired session records", deleted); + } + + Ok(deleted) + } + } + /// Find active collaboration sessions for a resource. /// /// Queries Constellation for session records referencing the given resource, @@ -1992,6 +2094,14 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { const SESSION_NSID: &str = "sh.weaver.collab.session"; + // Get authorized collaborators (owner is checked separately via URI authority) + let collaborators: std::collections::HashSet> = self + .find_collaborators_for_resource(resource_uri) + .await + .unwrap_or_default() + .into_iter() + .collect(); + let constellation_url = Url::parse(CONSTELLATION_URL).map_err(|e| { AgentError::from(ClientError::invalid_request(format!( "Invalid constellation URL: {}", @@ -2055,6 +2165,16 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { } } + // Check if peer is authorized (has valid invite+accept pair) + let peer_did = record_id.did.clone().into_static(); + if !collaborators.contains(&peer_did) { + tracing::debug!( + peer = %peer_did, + "Filtering out unauthorized session peer" + ); + continue; + } + peers.push(SessionPeer { did: record_id.did.into_static(), node_id: session_record.value.node_id.to_string(), diff --git a/crates/weaver-common/src/transport/messages.rs b/crates/weaver-common/src/transport/messages.rs index caea7ff..38d7a88 100644 --- a/crates/weaver-common/src/transport/messages.rs +++ b/crates/weaver-common/src/transport/messages.rs @@ -1,5 +1,6 @@ //! Wire protocol for collaborative editing messages. +use iroh::{PublicKey, SecretKey, Signature}; use serde::{Deserialize, Serialize}; /// Messages exchanged between collaborators over gossip. @@ -53,17 +54,89 @@ pub enum CollabMessage { } impl CollabMessage { - /// Serialize message to CBOR bytes for wire transmission + /// Serialize message to postcard bytes for wire transmission. pub fn to_bytes(&self) -> Result, postcard::Error> { postcard::to_stdvec(self) } - /// Deserialize message from CBOR bytes + /// Deserialize message from postcard bytes. pub fn from_bytes(bytes: &[u8]) -> Result { postcard::from_bytes(bytes) } } +/// A signed message wrapper for authenticated transport. +/// +/// Includes the sender's public key so receivers can verify without context. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SignedMessage { + /// Sender's public key (also their EndpointId). + pub from: PublicKey, + /// The serialized TimestampedMessage (postcard bytes). + pub data: Vec, + /// Ed25519 signature over data. + pub signature: Signature, +} + +/// Versioned wire format with timestamp. +#[derive(Debug, Clone, Serialize, Deserialize)] +enum WireMessage { + V0 { timestamp: u64, message: CollabMessage }, +} + +/// A verified message with sender and timestamp info. +#[derive(Debug, Clone)] +pub struct ReceivedMessage { + /// Sender's public key. + pub from: PublicKey, + /// When the message was sent (micros since epoch). + pub timestamp: u64, + /// The decoded message. + pub message: CollabMessage, +} + +/// Error type for signed message operations. +#[derive(Debug, thiserror::Error)] +pub enum SignedMessageError { + #[error("serialization failed: {0}")] + Serialization(#[from] postcard::Error), + #[error("signature verification failed")] + InvalidSignature, +} + +impl SignedMessage { + /// Sign a message and encode to bytes for wire transmission. + pub fn sign_and_encode(secret_key: &SecretKey, message: &CollabMessage) -> Result, SignedMessageError> { + use web_time::SystemTime; + + let timestamp = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_micros() as u64; + let wire = WireMessage::V0 { timestamp, message: message.clone() }; + let data = postcard::to_stdvec(&wire)?; + let signature = secret_key.sign(&data); + let from = secret_key.public(); + let signed = Self { from, data, signature }; + Ok(postcard::to_stdvec(&signed)?) + } + + /// Decode from bytes and verify signature. + pub fn decode_and_verify(bytes: &[u8]) -> Result { + let signed: Self = postcard::from_bytes(bytes)?; + signed.from + .verify(&signed.data, &signed.signature) + .map_err(|_| SignedMessageError::InvalidSignature)?; + let wire: WireMessage = postcard::from_bytes(&signed.data)?; + let WireMessage::V0 { timestamp, message } = wire; + Ok(ReceivedMessage { + from: signed.from, + timestamp, + message, + }) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/weaver-common/src/transport/mod.rs b/crates/weaver-common/src/transport/mod.rs index 41e0267..3a65568 100644 --- a/crates/weaver-common/src/transport/mod.rs +++ b/crates/weaver-common/src/transport/mod.rs @@ -14,10 +14,7 @@ mod session; pub use discovery::{node_id_to_string, parse_node_id, DiscoveredPeer, DiscoveryError}; pub use iroh::EndpointId; -pub use messages::CollabMessage; +pub use messages::{CollabMessage, ReceivedMessage, SignedMessage, SignedMessageError}; pub use node::{CollabNode, TransportError}; pub use presence::{Collaborator, PresenceTracker, RemoteCursor}; pub use session::{CollabSession, SessionError, SessionEvent, TopicId}; - -/// ALPN protocol identifier for weaver collaboration -pub const WEAVER_GOSSIP_ALPN: &[u8] = b"weaver/collab/0"; diff --git a/crates/weaver-common/src/transport/node.rs b/crates/weaver-common/src/transport/node.rs index 11ea7c1..1d1644c 100644 --- a/crates/weaver-common/src/transport/node.rs +++ b/crates/weaver-common/src/transport/node.rs @@ -3,12 +3,10 @@ use iroh::Endpoint; use iroh::EndpointId; use iroh::SecretKey; -use iroh_gossip::net::Gossip; +use iroh_gossip::net::{GOSSIP_ALPN, Gossip}; use miette::Diagnostic; use std::sync::Arc; -use super::WEAVER_GOSSIP_ALPN; - /// Error type for transport operations #[derive(Debug, thiserror::Error, Diagnostic)] #[diagnostic(code(weaver::transport))] @@ -48,7 +46,7 @@ impl CollabNode { // In native, this can do direct P2P with relay fallback let endpoint = Endpoint::builder() .secret_key(secret_key.clone()) - .alpns(vec![WEAVER_GOSSIP_ALPN.to_vec()]) + .alpns(vec![GOSSIP_ALPN.to_vec()]) .bind() .await .map_err(|e| TransportError::Bind(Box::new(e)))?; @@ -58,7 +56,7 @@ impl CollabNode { // Build router to dispatch incoming connections by ALPN let router = iroh::protocol::Router::builder(endpoint.clone()) - .accept(WEAVER_GOSSIP_ALPN, gossip.clone()) + .accept(GOSSIP_ALPN, gossip.clone()) .spawn(); tracing::info!(node_id = %endpoint.id(), "CollabNode started"); @@ -98,4 +96,52 @@ impl CollabNode { pub fn secret_key(&self) -> SecretKey { self.secret_key.clone() } + + /// Get the relay URL this node is connected to (if any). + /// + /// This should be published in session records so other peers can connect + /// via relay (essential for browser-to-browser connections). + pub fn relay_url(&self) -> Option { + self.endpoint + .addr() + .relay_urls() + .next() + .map(|url| url.to_string()) + } + + /// Get the full node address including relay info. + /// + /// Use this when you need to connect to this node from another peer. + pub fn node_addr(&self) -> iroh::EndpointAddr { + self.endpoint.addr() + } + + /// Wait for the endpoint to be online (relay connected). + /// + /// This should be called before publishing session records to ensure + /// the relay URL is available for peer discovery. For browser clients, + /// relay is required - we wait indefinitely since there's no fallback. + pub async fn wait_online(&self) { + self.endpoint.online().await; + } + + /// Wait for relay connection and return the relay URL. + /// + /// Waits indefinitely for relay - browser clients require relay URLs + /// for peer discovery. Returns the relay URL once connected. + pub async fn wait_for_relay(&self) -> String { + self.endpoint.online().await; + // After online(), relay_url should always be Some for browser clients + self.relay_url() + .expect("relay URL should be available after online()") + } + + /// Watch for address changes (including relay URL changes). + /// + /// Returns a stream that yields the address on each change. + /// Use this to detect relay URL changes and update session records. + pub fn watch_addr(&self) -> n0_future::boxed::BoxStream { + use iroh::Watcher; + Box::pin(self.endpoint.watch_addr().stream()) + } } diff --git a/crates/weaver-common/src/transport/session.rs b/crates/weaver-common/src/transport/session.rs index ae1dcda..f2b78fe 100644 --- a/crates/weaver-common/src/transport/session.rs +++ b/crates/weaver-common/src/transport/session.rs @@ -9,7 +9,7 @@ use n0_future::StreamExt; use n0_future::boxed::BoxStream; use n0_future::stream; -use super::{CollabMessage, CollabNode}; +use super::{CollabMessage, CollabNode, SignedMessage}; /// Topic ID for a gossip session - derived from resource URI. pub type TopicId = iroh_gossip::TopicId; @@ -57,7 +57,6 @@ pub enum SessionEvent { pub struct CollabSession { topic: TopicId, sender: GossipSender, - #[allow(dead_code)] node: Arc, } @@ -79,15 +78,27 @@ impl CollabSession { node: Arc, topic: TopicId, bootstrap_peers: Vec, - ) -> Result<(Self, BoxStream), SessionError> { + ) -> Result<(Self, BoxStream>), SessionError> { + tracing::info!( + topic = ?topic, + bootstrap_count = bootstrap_peers.len(), + "CollabSession: joining topic" + ); + + for peer in &bootstrap_peers { + tracing::debug!(peer = %peer, "CollabSession: bootstrap peer"); + } + // Subscribe to the gossip topic let (sender, receiver) = node .gossip() - .subscribe(topic, bootstrap_peers) + .subscribe_and_join(topic, bootstrap_peers) .await .map_err(|e| SessionError::Subscribe(Box::new(e)))? .split(); + tracing::info!("CollabSession: subscribed to gossip topic"); + let session = Self { topic, sender, @@ -101,49 +112,79 @@ impl CollabSession { } /// Convert gossip receiver into a stream of session events. - fn event_stream(receiver: GossipReceiver) -> BoxStream { - let stream = stream::unfold(receiver, |mut receiver| async move { + fn event_stream(receiver: GossipReceiver) -> BoxStream> { + let stream = stream::try_unfold(receiver, |mut receiver| async move { loop { - match receiver.next().await { - Some(Ok(event)) => { - let session_event = match event { - Event::NeighborUp(peer) => SessionEvent::PeerJoined(peer), - Event::NeighborDown(peer) => SessionEvent::PeerLeft(peer), - Event::Received(msg) => match CollabMessage::from_bytes(&msg.content) { - Ok(message) => SessionEvent::Message { - from: msg.delivered_from, - message, - }, - Err(e) => { - tracing::warn!(?e, "failed to decode collab message"); + let Some(event) = receiver.try_next().await.map_err(|e| { + tracing::error!(?e, "CollabSession: gossip receiver error"); + SessionError::Decode(Box::new(e)) + })? + else { + tracing::debug!("CollabSession: gossip stream ended"); + return Ok(None); + }; + + tracing::debug!(?event, "CollabSession: raw gossip event"); + let session_event = match event { + Event::NeighborUp(peer) => { + tracing::info!(peer = %peer, "CollabSession: neighbor up"); + SessionEvent::PeerJoined(peer) + } + Event::NeighborDown(peer) => { + tracing::info!(peer = %peer, "CollabSession: neighbor down"); + SessionEvent::PeerLeft(peer) + } + Event::Received(msg) => { + tracing::debug!( + from = %msg.delivered_from, + bytes = msg.content.len(), + "CollabSession: received message" + ); + match SignedMessage::decode_and_verify(&msg.content) { + Ok(received) => { + // Verify claimed sender matches transport sender + if received.from != msg.delivered_from { + tracing::warn!( + claimed = %received.from, + transport = %msg.delivered_from, + "sender mismatch - possible spoofing attempt" + ); continue; } - }, - Event::Lagged => { - tracing::warn!("gossip receiver lagged, some messages may be lost"); + SessionEvent::Message { + from: received.from, + message: received.message, + } + } + Err(e) => { + tracing::warn!(?e, "failed to verify/decode signed message"); continue; } - }; - return Some((session_event, receiver)); + } } - Some(Err(e)) => { - tracing::warn!(?e, "gossip receiver error"); + Event::Lagged => { + tracing::warn!("gossip receiver lagged, some messages may be lost"); continue; } - None => return None, - } + }; + break Ok(Some((session_event, receiver))); } }); Box::pin(stream) } - /// Broadcast a message to all peers in the session. + /// Broadcast a signed message to all peers in the session. pub async fn broadcast(&self, message: &CollabMessage) -> Result<(), SessionError> { - let bytes = message - .to_bytes() + let bytes = SignedMessage::sign_and_encode(&self.node.secret_key(), message) .map_err(|e| SessionError::Broadcast(Box::new(e)))?; + tracing::debug!( + bytes = bytes.len(), + topic = ?self.topic, + "CollabSession: broadcasting signed message" + ); + self.sender .broadcast(bytes.into()) .await @@ -156,4 +197,26 @@ impl CollabSession { pub fn topic(&self) -> TopicId { self.topic } + + /// Add new peers to the gossip session. + /// + /// Use this to add peers discovered after initial subscription. + /// The gossip layer will attempt to connect to these peers. + pub async fn join_peers(&self, peers: Vec) -> Result<(), SessionError> { + if peers.is_empty() { + return Ok(()); + } + tracing::info!( + count = peers.len(), + "CollabSession: joining additional peers" + ); + for peer in &peers { + tracing::debug!(peer = %peer, "CollabSession: adding peer"); + } + self.sender + .join_peers(peers) + .await + .map_err(|e| SessionError::Subscribe(Box::new(e)))?; + Ok(()) + } } -- 2.51.2