feat(presence): live friends listener + ownership move (W7 B2)
Makes the friends list live, building on the B1 persistent endpoint.
Friends ownership moves into core (was the GUI's):
- Core loads/owns friends.json behind a shared Mutex<FriendStore>; a malformed
load yields an empty store flagged READ-ONLY so we never overwrite the damaged
file (fixes backlog A16). New commands AddFriend/RemoveFriend/RenameFriend +
UiEvent::FriendsUpdated{friends,read_only}; the GUI is now a read-only mirror
that renders from the event and drives mutations via commands. The friends UI
shows a warning + blocks edits when read-only.
- Presence posture pushed to core via SetPresenceMode (persistence stays in
AppConfig); held in a shared Mutex for the listener/scheduler.
Live listener + outbound scheduler:
- New FriendsProtocol ProtocolHandler on the persistent Router for FRIENDS_ALPN
(the router owns accept(), so the listener can't be presence_net::serve — same
delegation pattern as B1's AudioRouter). Its reply policy reads the shared
friends/mode/current-room and uses presence::should_answer: answer friends only,
never while invisible, and report our current gathering's restamped member
ticket so a friend can one-click Join. handle()'s body is factored into a shared
exchange() used by both serve (tests) and FriendsProtocol.
- Outbound ping scheduler folded into the core loop via tokio::select! on a slow
interval (60s, first pass delayed 3s for endpoint online). FULLY DARK while
Invisible (no probing at all — user's choice). Each pass runs detached so it
never blocks command handling and picks up a rebuilt stack next tick; probes
friends with a saved addr in parallel and emits UiEvent::FriendPresence.
- note_seen auto-heal: a connected peer who is a friend has their last_addr
refreshed (+persisted) so the scheduler can reach them later.
- current_room shared state set on Join (restamped ticket) / cleared on Leave.
P5 UI: each friend shows online / offline / in-room with a one-click Join.
256 lib + 6 reconnect + 4 loopback + 2 ignored real-endpoint tests green, clippy
--all-targets clean, release builds. B2a (ownership/A16) is solo-verifiable; the
live listener + scheduler need the 2-machine field test before this merges.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
+110
-32
@@ -146,6 +146,9 @@ pub enum AppMessage {
|
||||
AddFriend,
|
||||
RemoveFriend(EndpointId),
|
||||
RenameFriend(EndpointId, String),
|
||||
/// Join the gathering a friend is in (W7 B2), via the member ticket their
|
||||
/// presence reply carried. Mirrors a manual ticket join.
|
||||
JoinFriendRoom(String),
|
||||
EventOccurred(Event),
|
||||
NavigateToSettings,
|
||||
NavigateBack,
|
||||
@@ -278,8 +281,16 @@ pub struct AppState {
|
||||
identity_error: Option<String>,
|
||||
/// Whether the "Regenerate identity?" confirm modal is open.
|
||||
regenerate_identity_confirm_open: bool,
|
||||
/// Saved friends (W7 P2/P5), loaded at startup, persisted on every change.
|
||||
/// Saved friends (W7) — a READ-ONLY MIRROR of the core-owned store, refreshed by
|
||||
/// `UiEvent::FriendsUpdated`. The GUI no longer loads/saves it; add/remove/rename
|
||||
/// go to core as commands.
|
||||
friends: crate::friends::FriendStore,
|
||||
/// True when core couldn't load `friends.json` (malformed) and is in a degraded
|
||||
/// read-only state — the GUI disables edits + warns so we don't clobber it (A16).
|
||||
friends_read_only: bool,
|
||||
/// Latest live presence per friend (W7 B2), from `UiEvent::FriendPresence`. A
|
||||
/// missing entry = treat as offline/unknown.
|
||||
friend_presence: std::collections::HashMap<EndpointId, crate::presence::FriendPresence>,
|
||||
/// "Add friend" form inputs: their node id (hex) and an optional name.
|
||||
friend_add_id: String,
|
||||
friend_add_name: String,
|
||||
@@ -329,6 +340,7 @@ impl Default for AppState {
|
||||
let _ = controller.send(CoreCommand::SetNetworkMode(config.network_mode));
|
||||
let _ = controller.send(CoreCommand::SetRecordingMode(config.recording_mode));
|
||||
let _ = controller.send(CoreCommand::SetPixelpassPath(config.pixelpass_path.clone()));
|
||||
let _ = controller.send(CoreCommand::SetPresenceMode(config.presence_mode));
|
||||
let pixelpass_available =
|
||||
crate::screenshare::is_available(config.pixelpass_path.as_deref());
|
||||
let all_devices = enumerate_audio_devices();
|
||||
@@ -382,10 +394,11 @@ impl Default for AppState {
|
||||
identity_persisted: true,
|
||||
identity_error: None,
|
||||
regenerate_identity_confirm_open: false,
|
||||
friends: crate::friends::load().unwrap_or_else(|e| {
|
||||
crate::log_msg(&format!("friends: failed to load, starting empty: {e:#}"));
|
||||
crate::friends::FriendStore::default()
|
||||
}),
|
||||
// Core owns the friends store now; the GUI starts empty and fills in
|
||||
// from the FriendsUpdated event core emits at startup.
|
||||
friends: crate::friends::FriendStore::default(),
|
||||
friends_read_only: false,
|
||||
friend_presence: std::collections::HashMap::new(),
|
||||
friend_add_id: String::new(),
|
||||
friend_add_name: String::new(),
|
||||
friend_add_error: None,
|
||||
@@ -683,6 +696,17 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
state.identity_persisted = persisted;
|
||||
state.identity_error = error;
|
||||
}
|
||||
UiEvent::FriendsUpdated { friends, read_only } => {
|
||||
// Core owns the store; mirror its snapshot. Drop presence for
|
||||
// anyone no longer a friend so the UI doesn't show a stale dot.
|
||||
let ids: HashSet<EndpointId> = friends.iter().map(|f| f.id).collect();
|
||||
state.friend_presence.retain(|id, _| ids.contains(id));
|
||||
state.friends.friends = friends;
|
||||
state.friends_read_only = read_only;
|
||||
}
|
||||
UiEvent::FriendPresence { id, presence } => {
|
||||
state.friend_presence.insert(id, presence);
|
||||
}
|
||||
UiEvent::Error(err) => {
|
||||
state.status_message = format!("Error: {}", err);
|
||||
}
|
||||
@@ -764,8 +788,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
AppMessage::PresenceModeSelected(mode) => {
|
||||
state.config.presence_mode = mode;
|
||||
state.config.save();
|
||||
// Persisted now; the live friends listener (P4, not yet wired) reads
|
||||
// this posture when it lands. No core command until then.
|
||||
// Push to core, which gates the live listener + ping scheduler (B2).
|
||||
let _ = state.controller.send(CoreCommand::SetPresenceMode(mode));
|
||||
}
|
||||
AppMessage::FriendAddIdChanged(val) => {
|
||||
state.friend_add_id = val;
|
||||
@@ -776,41 +800,60 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
state.friend_add_error = None;
|
||||
}
|
||||
AppMessage::AddFriend => {
|
||||
let id_str = state.friend_add_id.trim();
|
||||
match id_str.parse::<EndpointId>() {
|
||||
Ok(id) if state.friends.contains(&id) => {
|
||||
state.friend_add_error = Some("Already in your friends list.".to_string());
|
||||
}
|
||||
Ok(id) => {
|
||||
let name = {
|
||||
let n = state.friend_add_name.trim();
|
||||
if n.is_empty() { short_id(id_str) } else { n.to_string() }
|
||||
};
|
||||
state.friends.add(id, name, None);
|
||||
if let Err(e) = crate::friends::save(&state.friends) {
|
||||
crate::log_msg(&format!("friends: save failed: {e:#}"));
|
||||
// Core owns the store; the GUI just validates the id locally, then sends
|
||||
// a command. The mirror (and the cleared form) update on FriendsUpdated.
|
||||
if state.friends_read_only {
|
||||
state.friend_add_error =
|
||||
Some("Friends list is read-only (couldn't load friends.json).".to_string());
|
||||
} else {
|
||||
let id_str = state.friend_add_id.trim();
|
||||
match id_str.parse::<EndpointId>() {
|
||||
Ok(id) if state.friends.contains(&id) => {
|
||||
state.friend_add_error = Some("Already in your friends list.".to_string());
|
||||
}
|
||||
Ok(id) => {
|
||||
let name = {
|
||||
let n = state.friend_add_name.trim();
|
||||
if n.is_empty() { short_id(id_str) } else { n.to_string() }
|
||||
};
|
||||
let _ = state.controller.send(CoreCommand::AddFriend { id, name, addr: None });
|
||||
state.friend_add_id.clear();
|
||||
state.friend_add_name.clear();
|
||||
state.friend_add_error = None;
|
||||
}
|
||||
Err(_) => {
|
||||
state.friend_add_error = Some("That doesn't look like a valid node ID.".to_string());
|
||||
}
|
||||
state.friend_add_id.clear();
|
||||
state.friend_add_name.clear();
|
||||
state.friend_add_error = None;
|
||||
}
|
||||
Err(_) => {
|
||||
state.friend_add_error = Some("That doesn't look like a valid node ID.".to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
AppMessage::RemoveFriend(id) => {
|
||||
state.friends.remove(&id);
|
||||
if let Err(e) = crate::friends::save(&state.friends) {
|
||||
crate::log_msg(&format!("friends: save failed: {e:#}"));
|
||||
if !state.friends_read_only {
|
||||
let _ = state.controller.send(CoreCommand::RemoveFriend(id));
|
||||
}
|
||||
}
|
||||
AppMessage::RenameFriend(id, new_name) => {
|
||||
state.friends.rename(&id, new_name);
|
||||
if let Err(e) = crate::friends::save(&state.friends) {
|
||||
crate::log_msg(&format!("friends: save failed: {e:#}"));
|
||||
if !state.friends_read_only {
|
||||
let _ = state.controller.send(CoreCommand::RenameFriend(id, new_name));
|
||||
}
|
||||
}
|
||||
AppMessage::JoinFriendRoom(ticket) => {
|
||||
// Join via the friend's member ticket (same path as a manual join).
|
||||
let input_device = state.selected_input.as_ref().map(|d| d.name.clone());
|
||||
let output_device = state.selected_output.as_ref().map(|d| d.name.clone());
|
||||
state.status_message = "Joining your friend's room...".to_string();
|
||||
state.config.username = state.name.clone();
|
||||
state.config.save();
|
||||
state.mic_test_active = false;
|
||||
let _ = state.controller.send(CoreCommand::Join {
|
||||
name: state.name.clone(),
|
||||
ticket,
|
||||
input_device,
|
||||
output_device,
|
||||
echo_cancellation: state.config.echo_cancellation_enabled,
|
||||
avatar: state.config.avatar.clone(),
|
||||
});
|
||||
}
|
||||
AppMessage::ToggleNotifications(enabled) => {
|
||||
state.config.notifications_enabled = enabled;
|
||||
state.config.save();
|
||||
@@ -1626,6 +1669,28 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
for f in state.friends.list() {
|
||||
let fid = f.id;
|
||||
let id_short = format!("{}…", short_id(&f.id.to_string()));
|
||||
// Live presence (W7 B2), from the latest FriendPresence event: a
|
||||
// missing entry = offline. An in-room friend gets a one-click Join.
|
||||
let status: Element<AppMessage> = match state.friend_presence.get(&fid) {
|
||||
Some(crate::presence::FriendPresence::InRoom { name, ticket }) => {
|
||||
let label =
|
||||
if name.is_empty() { "in a room".to_string() } else { format!("in {name}") };
|
||||
row![
|
||||
text(label).size(11).color(color_green),
|
||||
button(text("Join").size(12))
|
||||
.on_press(AppMessage::JoinFriendRoom(ticket.clone()))
|
||||
.style(b_style(color_surface, color_blue, color_text, 6.0))
|
||||
.padding(6),
|
||||
]
|
||||
.spacing(6)
|
||||
.align_y(iced::alignment::Vertical::Center)
|
||||
.into()
|
||||
}
|
||||
Some(crate::presence::FriendPresence::Online) => {
|
||||
text("● online").size(11).color(color_green).into()
|
||||
}
|
||||
None => text("○ offline").size(11).color(color_subtext).into(),
|
||||
};
|
||||
friend_rows = friend_rows.push(
|
||||
row![
|
||||
text_input("name", &f.name)
|
||||
@@ -1638,6 +1703,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
.style(b_style(color_surface, color_maroon, color_text, 6.0))
|
||||
.padding(6),
|
||||
text(id_short).size(11).color(color_subtext),
|
||||
status,
|
||||
// Absorb the remaining width so the row is left-aligned and
|
||||
// nothing stretches to / under the scrollbar edge.
|
||||
horizontal_space(),
|
||||
@@ -1651,11 +1717,23 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
|
||||
Some(e) => text(e).size(11).color(color_red).into(),
|
||||
None => column![].into(),
|
||||
};
|
||||
// A16: if core couldn't load friends.json it runs read-only so it never
|
||||
// overwrites the damaged file — tell the user, since edits won't stick.
|
||||
let friends_readonly_warning: Element<AppMessage> = if state.friends_read_only {
|
||||
text("⚠ Couldn't load friends.json — the list is read-only so the damaged \
|
||||
file isn't overwritten. Fix or remove it, then restart.")
|
||||
.size(12)
|
||||
.color(color_red)
|
||||
.into()
|
||||
} else {
|
||||
column![].into()
|
||||
};
|
||||
let friends_section = column![
|
||||
text("Save people by their node ID — this is the durable part; you \
|
||||
recognise friends here no matter what room you're in.")
|
||||
.size(12)
|
||||
.color(color_subtext),
|
||||
friends_readonly_warning,
|
||||
friend_rows,
|
||||
row![
|
||||
text_input("Friend's node ID", &state.friend_add_id)
|
||||
|
||||
+25
-1
@@ -1,6 +1,8 @@
|
||||
use crate::config::{NetworkMode, RecordingMode};
|
||||
use crate::friends::Friend;
|
||||
use crate::network::PeerState;
|
||||
use iroh::EndpointId;
|
||||
use crate::presence::{FriendPresence, PresenceMode};
|
||||
use iroh::{EndpointAddr, EndpointId};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum CoreCommand {
|
||||
@@ -52,6 +54,18 @@ pub enum CoreCommand {
|
||||
/// on the next room join (the endpoint is rebuilt then). The core replies with
|
||||
/// an updated [`UiEvent::IdentityStatus`].
|
||||
RegenerateIdentity,
|
||||
/// Add a friend (W7). Core owns the friends store: it mutates + persists it and
|
||||
/// replies with [`UiEvent::FriendsUpdated`]. `addr` seeds `last_addr` if known
|
||||
/// (e.g. added from a room). Idempotent — re-adding an existing id is a no-op.
|
||||
AddFriend { id: EndpointId, name: String, addr: Option<EndpointAddr> },
|
||||
/// Remove a friend by id (W7).
|
||||
RemoveFriend(EndpointId),
|
||||
/// Locally rename a friend (W7).
|
||||
RenameFriend(EndpointId, String),
|
||||
/// Set our presence posture (W7). Gates the idle listener (answer friends-only /
|
||||
/// invisible) and the outbound ping scheduler (invisible = fully dark). Sent at
|
||||
/// startup from config and whenever the user changes it.
|
||||
SetPresenceMode(PresenceMode),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -89,5 +103,15 @@ pub enum UiEvent {
|
||||
/// since the id (and thus friend recognition) won't survive the next launch.
|
||||
/// `error` carries the reason when degraded, for the UI explainer.
|
||||
IdentityStatus { node_id: String, persisted: bool, error: Option<String> },
|
||||
/// The friends list (W7), now owned by core. Sent at startup (after load) and
|
||||
/// after every add/remove/rename so the GUI renders from this snapshot instead
|
||||
/// of owning the store. `read_only` is true when `friends.json` failed to load
|
||||
/// (malformed) — the GUI shows a degraded warning and disables edits so we never
|
||||
/// overwrite the damaged file (backlog A16).
|
||||
FriendsUpdated { friends: Vec<Friend>, read_only: bool },
|
||||
/// A friend's live presence from a successful ping reply (W7): online, or in a
|
||||
/// joinable gathering (with a one-click ticket). Emitted by the outbound ping
|
||||
/// scheduler; absence of a recent event = treat as offline.
|
||||
FriendPresence { id: EndpointId, presence: FriendPresence },
|
||||
Error(String),
|
||||
}
|
||||
|
||||
+223
-6
@@ -414,6 +414,7 @@ async fn build_net_stack(
|
||||
secret_key: SecretKey,
|
||||
network_mode: NetworkMode,
|
||||
memory_lookup: iroh::address_lookup::memory::MemoryLookup,
|
||||
friends_handler: crate::presence_net::Handler,
|
||||
) -> Result<NetStack, anyhow::Error> {
|
||||
// Build the endpoint per the configured relay/discovery posture. All postures
|
||||
// keep the in-memory address lookup (fed by tickets and gossip); they differ in
|
||||
@@ -458,9 +459,17 @@ async fn build_net_stack(
|
||||
.spawn(endpoint.clone());
|
||||
|
||||
let audio_router = AudioRouter::new();
|
||||
// The friends presence listener (W7 B2) rides this same persistent router as a
|
||||
// third ALPN — it MUST be a handler here, not a standalone accept loop, since
|
||||
// the router owns endpoint.accept(). Policy (who we answer / what room we
|
||||
// report) is injected via `friends_handler`.
|
||||
let router = Router::builder(endpoint.clone())
|
||||
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
|
||||
.accept(b"peerspeak-audio", audio_router.clone())
|
||||
.accept(
|
||||
crate::presence_net::FRIENDS_ALPN,
|
||||
crate::presence_net::FriendsProtocol::new(friends_handler),
|
||||
)
|
||||
.spawn();
|
||||
|
||||
Ok(NetStack {
|
||||
@@ -507,6 +516,63 @@ async fn stop_recording(
|
||||
}
|
||||
}
|
||||
|
||||
/// Persist the (core-owned) friends store and tell the GUI the new snapshot. Skips
|
||||
/// the disk write when `read_only` (a malformed load — A16: don't clobber it) but
|
||||
/// still emits so the UI reflects the in-memory change. Snapshots under the lock,
|
||||
/// then releases it before the async send.
|
||||
async fn persist_and_emit_friends(
|
||||
friends: &Arc<std::sync::Mutex<crate::friends::FriendStore>>,
|
||||
read_only: bool,
|
||||
ui_tx: &mpsc::Sender<UiEvent>,
|
||||
) {
|
||||
let store = friends.lock().unwrap().clone();
|
||||
if !read_only
|
||||
&& let Err(e) = crate::friends::save(&store)
|
||||
{
|
||||
crate::log_msg(&format!("friends: save failed: {e:#}"));
|
||||
}
|
||||
let _ = ui_tx
|
||||
.send(UiEvent::FriendsUpdated { friends: store.friends, read_only })
|
||||
.await;
|
||||
}
|
||||
|
||||
/// How often the outbound presence scheduler refreshes friends' status. Slow on
|
||||
/// purpose — presence is best-effort, not real-time, and each pass opens a short
|
||||
/// connection per friend.
|
||||
const PING_INTERVAL: Duration = Duration::from_secs(60);
|
||||
/// Delay before the FIRST presence pass, so the endpoint's background `online()`
|
||||
/// has a moment to finish (otherwise the first probes fail and friends flash offline).
|
||||
const PING_STARTUP_DELAY: Duration = Duration::from_secs(3);
|
||||
|
||||
/// One outbound presence-refresh pass (W7 B2): probe every friend that has a saved
|
||||
/// address and emit their interpreted status. Friends with no saved address are
|
||||
/// skipped (a bare id can't resolve without discovery) and stay offline in the UI
|
||||
/// until first contact populates their address via `note_seen`. Probes run in
|
||||
/// parallel (friend counts are small); an unreachable friend just yields nothing.
|
||||
async fn probe_friends_once(
|
||||
endpoint: Endpoint,
|
||||
friends: crate::friends::FriendStore,
|
||||
ui_tx: mpsc::Sender<UiEvent>,
|
||||
) {
|
||||
let mut set = tokio::task::JoinSet::new();
|
||||
for f in friends.list() {
|
||||
let Some(addr) = f.last_addr.clone() else { continue };
|
||||
let id = f.id;
|
||||
let ep = endpoint.clone();
|
||||
set.spawn(async move {
|
||||
match crate::presence_net::probe(&ep, addr).await {
|
||||
Ok(reply) => crate::presence::interpret_pong(&reply).map(|p| (id, p)),
|
||||
Err(_) => None,
|
||||
}
|
||||
});
|
||||
}
|
||||
while let Some(res) = set.join_next().await {
|
||||
if let Ok(Some((id, presence))) = res {
|
||||
let _ = ui_tx.send(UiEvent::FriendPresence { id, presence }).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_core_loop(
|
||||
mut cmd_rx: mpsc::Receiver<CoreCommand>,
|
||||
ui_tx: mpsc::Sender<UiEvent>,
|
||||
@@ -588,12 +654,55 @@ async fn run_core_loop(
|
||||
// Standalone capture-only mic meter, live only when no session exists.
|
||||
let mut mic_monitor: Option<MicMonitor> = None;
|
||||
|
||||
// Friends store (W7) — core now OWNS it (was the GUI's). Shared so the idle
|
||||
// listener + ping scheduler read it. A malformed load yields an EMPTY store
|
||||
// flagged read-only, so we never save over the damaged file (A16); the GUI
|
||||
// surfaces the degraded state from `FriendsUpdated { read_only: true }`.
|
||||
let (initial_friends, friends_read_only) = match crate::friends::load() {
|
||||
Ok(store) => (store, false),
|
||||
Err(e) => {
|
||||
crate::log_msg(&format!(
|
||||
"friends: load failed, starting read-only to avoid clobbering the file: {e:#}"
|
||||
));
|
||||
(crate::friends::FriendStore::default(), true)
|
||||
}
|
||||
};
|
||||
let friends = Arc::new(std::sync::Mutex::new(initial_friends));
|
||||
// Our presence posture, shared with the listener + scheduler. The GUI pushes it
|
||||
// at startup + on change via SetPresenceMode; persistence stays in AppConfig.
|
||||
let presence_mode = Arc::new(std::sync::Mutex::new(crate::presence::PresenceMode::default()));
|
||||
// The gathering we're currently in (its restamped member ticket + label), so the
|
||||
// listener can offer friends a one-click Join. `None` when not in a call. Set on
|
||||
// Join, cleared on Leave.
|
||||
let current_room: Arc<std::sync::Mutex<Option<crate::presence::RoomPresence>>> =
|
||||
Arc::new(std::sync::Mutex::new(None));
|
||||
|
||||
// Reply policy for the idle friends listener (B2): answer friends only, never
|
||||
// while invisible (`should_answer`), and report our current gathering so a friend
|
||||
// can one-click join. Reads the shared snapshots, so it stays correct as they
|
||||
// change and survives a network-stack rebuild. Pure-sync (no awaits, no lock held
|
||||
// across one). Built once and handed to every `build_net_stack`.
|
||||
let friends_handler: crate::presence_net::Handler = {
|
||||
let friends = friends.clone();
|
||||
let presence_mode = presence_mode.clone();
|
||||
let current_room = current_room.clone();
|
||||
Arc::new(move |from| {
|
||||
let mode = *presence_mode.lock().unwrap();
|
||||
let allowed = crate::presence::should_answer(&from, &friends.lock().unwrap(), mode);
|
||||
if !allowed {
|
||||
return None;
|
||||
}
|
||||
let room = current_room.lock().unwrap().clone();
|
||||
Some(crate::presence::ControlMsg::Pong { room })
|
||||
})
|
||||
};
|
||||
|
||||
// The persistent network stack (endpoint + gossip + router), built once at
|
||||
// startup and kept alive for the app's lifetime. Room sessions ride on top of
|
||||
// it (subscribe a topic + bind the audio router on join, clear on leave); it's
|
||||
// rebuilt only when the network mode or identity changes. Moving `memory_lookup`
|
||||
// in — all later access is via `net.memory_lookup`.
|
||||
let mut net = match build_net_stack(secret_key.clone(), network_mode, memory_lookup).await {
|
||||
let mut net = match build_net_stack(secret_key.clone(), network_mode, memory_lookup, friends_handler.clone()).await {
|
||||
Ok(stack) => stack,
|
||||
Err(e) => {
|
||||
// Only a local socket bind can fail here (the relay handshake is
|
||||
@@ -610,7 +719,46 @@ async fn run_core_loop(
|
||||
// "applies on next join" semantics while keeping the endpoint up while idle.
|
||||
let mut net_rebuild_pending = false;
|
||||
|
||||
while let Some(cmd) = cmd_rx.recv().await {
|
||||
// Tell the GUI the loaded friends list (it renders from this, no longer owning
|
||||
// it). Snapshot under the lock, then release it before the async send.
|
||||
let initial_snapshot = friends.lock().unwrap().list().to_vec();
|
||||
let _ = ui_tx
|
||||
.send(UiEvent::FriendsUpdated {
|
||||
friends: initial_snapshot,
|
||||
read_only: friends_read_only,
|
||||
})
|
||||
.await;
|
||||
|
||||
// Drive the command loop AND the outbound presence ping scheduler together: the
|
||||
// scheduler can't block command handling, so a probe pass runs in a detached task
|
||||
// on each tick. `interval_at` delays the first pass so the endpoint can come
|
||||
// online first.
|
||||
let mut ping_interval = tokio::time::interval_at(
|
||||
tokio::time::Instant::now() + PING_STARTUP_DELAY,
|
||||
PING_INTERVAL,
|
||||
);
|
||||
ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
loop {
|
||||
let cmd = tokio::select! {
|
||||
maybe_cmd = cmd_rx.recv() => match maybe_cmd {
|
||||
Some(cmd) => cmd,
|
||||
None => break,
|
||||
},
|
||||
_ = ping_interval.tick() => {
|
||||
// Fully dark while Invisible (the user's choice): don't even probe,
|
||||
// so nothing we do touches a friend's machine. Otherwise refresh in a
|
||||
// detached task capturing the CURRENT endpoint (a stack rebuild between
|
||||
// ticks is naturally picked up next tick).
|
||||
if *presence_mode.lock().unwrap() != crate::presence::PresenceMode::Invisible {
|
||||
tokio::spawn(probe_friends_once(
|
||||
net.endpoint.clone(),
|
||||
friends.lock().unwrap().clone(),
|
||||
ui_tx.clone(),
|
||||
));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
};
|
||||
match cmd {
|
||||
CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation, avatar } => {
|
||||
current_name = name.clone();
|
||||
@@ -634,7 +782,7 @@ async fn run_core_loop(
|
||||
if net_rebuild_pending {
|
||||
let lookup = net.memory_lookup.clone();
|
||||
net.shutdown().await;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup).await?;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone()).await?;
|
||||
net_rebuild_pending = false;
|
||||
}
|
||||
|
||||
@@ -1049,6 +1197,11 @@ async fn run_core_loop(
|
||||
// The ticket of the room this event loop serves, so peer add/remove
|
||||
// updates the right per-ticket bucket in `known_peers` (A8 archive).
|
||||
let ticket_events = ticket_str.clone();
|
||||
// Friends store + ui sender, so a connected peer who is a friend has
|
||||
// their saved address auto-healed (W7) — populates `last_addr` so the
|
||||
// presence scheduler can reach them later.
|
||||
let friends_events = friends.clone();
|
||||
let friends_read_only_events = friends_read_only;
|
||||
let event_task = tokio::spawn(async move {
|
||||
while let Some(event) = room_events.recv().await {
|
||||
match event {
|
||||
@@ -1061,6 +1214,21 @@ async fn run_core_loop(
|
||||
// Hand over the full address so reconnects can dial
|
||||
// it directly rather than via the gossip lookup.
|
||||
transport_events.connect_peer(state.addr.clone()).await;
|
||||
// Auto-heal a friend's saved address (W7): if this
|
||||
// peer is a friend, refresh their last_addr so the
|
||||
// presence scheduler can reach them between rooms.
|
||||
let healed = friends_events
|
||||
.lock()
|
||||
.unwrap()
|
||||
.note_seen(&peer_id, state.addr.clone());
|
||||
if healed {
|
||||
persist_and_emit_friends(
|
||||
&friends_events,
|
||||
friends_read_only_events,
|
||||
&ui_tx_events,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
// Retain this peer under this room's ticket as a
|
||||
// future rejoin bootstrap target (A8).
|
||||
known_peers_events
|
||||
@@ -1105,6 +1273,21 @@ async fn run_core_loop(
|
||||
// re-records the same address.
|
||||
cancel_grace_timer(&grace_timers_events, &peer_id);
|
||||
transport_events.connect_peer(state.addr.clone()).await;
|
||||
// Auto-heal a friend's saved address (W7) on the
|
||||
// re-announce too — this is the path that catches a
|
||||
// friend who moved networks mid-session.
|
||||
let healed = friends_events
|
||||
.lock()
|
||||
.unwrap()
|
||||
.note_seen(&peer_id, state.addr.clone());
|
||||
if healed {
|
||||
persist_and_emit_friends(
|
||||
&friends_events,
|
||||
friends_read_only_events,
|
||||
&ui_tx_events,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
// Refresh this room's retained rejoin target with the
|
||||
// fresh addr (A8).
|
||||
known_peers_events
|
||||
@@ -1192,6 +1375,13 @@ async fn run_core_loop(
|
||||
// creator (same addr+topic), and a no-op if the ticket can't be
|
||||
// parsed (a malformed join, which fails anyway).
|
||||
let share_ticket = PeerSpeakTicket::restamp(&ticket_str, endpoint.addr());
|
||||
// Advertise this gathering to friends who ping us (W7 B2): our own
|
||||
// restamped member ticket → a one-click Join. Name is empty until
|
||||
// cosmetic room labels exist (P5/recents).
|
||||
*current_room.lock().unwrap() = Some(crate::presence::RoomPresence {
|
||||
name: String::new(),
|
||||
ticket: share_ticket.clone(),
|
||||
});
|
||||
let _ = ui_tx.send(UiEvent::RoomJoined { ticket: share_ticket, self_id }).await;
|
||||
active_session = Some(session);
|
||||
}
|
||||
@@ -1204,6 +1394,8 @@ async fn run_core_loop(
|
||||
session.shutdown(audio_backend.clone()).await;
|
||||
// Stop routing inbound audio links — the endpoint/router stay up.
|
||||
net.audio_router.clear();
|
||||
// No longer in a gathering — friends who ping see us as just online.
|
||||
*current_room.lock().unwrap() = None;
|
||||
let _ = ui_tx.send(UiEvent::RoomLeft).await;
|
||||
}
|
||||
// Apply any network-mode / identity change that was deferred while we
|
||||
@@ -1211,7 +1403,7 @@ async fn run_core_loop(
|
||||
if net_rebuild_pending {
|
||||
let lookup = net.memory_lookup.clone();
|
||||
net.shutdown().await;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup).await?;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone()).await?;
|
||||
net_rebuild_pending = false;
|
||||
}
|
||||
}
|
||||
@@ -1324,7 +1516,7 @@ async fn run_core_loop(
|
||||
if active_session.is_none() {
|
||||
let lookup = net.memory_lookup.clone();
|
||||
net.shutdown().await;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup).await?;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone()).await?;
|
||||
} else {
|
||||
net_rebuild_pending = true;
|
||||
}
|
||||
@@ -1356,7 +1548,7 @@ async fn run_core_loop(
|
||||
if active_session.is_none() {
|
||||
let lookup = net.memory_lookup.clone();
|
||||
net.shutdown().await;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup).await?;
|
||||
net = build_net_stack(secret_key.clone(), network_mode, lookup, friends_handler.clone()).await?;
|
||||
} else {
|
||||
net_rebuild_pending = true;
|
||||
}
|
||||
@@ -1370,6 +1562,31 @@ async fn run_core_loop(
|
||||
.await;
|
||||
}
|
||||
|
||||
CoreCommand::AddFriend { id, name, addr } => {
|
||||
// Idempotent: re-adding an existing id is a no-op (preserves the
|
||||
// local name/addr), so only persist+emit when something changed.
|
||||
let changed = friends.lock().unwrap().add(id, name, addr);
|
||||
if changed {
|
||||
persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
|
||||
}
|
||||
}
|
||||
|
||||
CoreCommand::RemoveFriend(id) => {
|
||||
if friends.lock().unwrap().remove(&id) {
|
||||
persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
|
||||
}
|
||||
}
|
||||
|
||||
CoreCommand::RenameFriend(id, new_name) => {
|
||||
if friends.lock().unwrap().rename(&id, new_name) {
|
||||
persist_and_emit_friends(&friends, friends_read_only, &ui_tx).await;
|
||||
}
|
||||
}
|
||||
|
||||
CoreCommand::SetPresenceMode(mode) => {
|
||||
*presence_mode.lock().unwrap() = mode;
|
||||
}
|
||||
|
||||
CoreCommand::SetRecordingMode(mode) => {
|
||||
recording_mode = mode;
|
||||
}
|
||||
|
||||
@@ -101,6 +101,13 @@ pub async fn serve(endpoint: Endpoint, handler: Handler) {
|
||||
|
||||
async fn handle(incoming: Incoming, handler: Handler) -> Result<()> {
|
||||
let conn = incoming.await.context("inbound connection failed")?;
|
||||
exchange(&conn, &handler).await
|
||||
}
|
||||
|
||||
/// One ping→pong exchange on an already-accepted connection: read the ping,
|
||||
/// ask the policy, reply (or reveal nothing), close. Shared by the standalone
|
||||
/// [`serve`] loop and the [`FriendsProtocol`] router handler.
|
||||
async fn exchange(conn: &iroh::endpoint::Connection, handler: &Handler) -> Result<()> {
|
||||
// The authenticated remote id — NOT anything the peer puts in the payload.
|
||||
let from = conn.remote_id();
|
||||
|
||||
@@ -129,6 +136,48 @@ async fn handle(incoming: Incoming, handler: Handler) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The live friends-presence listener as an iroh [`ProtocolHandler`], registered
|
||||
/// once on the app's single persistent `Router` for [`FRIENDS_ALPN`]. Because the
|
||||
/// Router owns `endpoint.accept()`, the listener can't be the standalone [`serve`]
|
||||
/// loop (that would compete for accepts); this delegates each inbound connection to
|
||||
/// the same [`exchange`] body, with the reply policy injected as a [`Handler`]
|
||||
/// (which wraps [`crate::presence::should_answer`] + builds the Pong). Mirrors the
|
||||
/// `AudioRouter` pattern from the B1 persistent-endpoint refactor.
|
||||
#[derive(Clone)]
|
||||
pub struct FriendsProtocol {
|
||||
handler: Handler,
|
||||
}
|
||||
|
||||
impl FriendsProtocol {
|
||||
pub fn new(handler: Handler) -> Self {
|
||||
Self { handler }
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for FriendsProtocol {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("FriendsProtocol").finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl iroh::protocol::ProtocolHandler for FriendsProtocol {
|
||||
fn accept(
|
||||
&self,
|
||||
connection: iroh::endpoint::Connection,
|
||||
) -> impl std::future::Future<Output = Result<(), iroh::protocol::AcceptError>> + Send {
|
||||
let handler = self.handler.clone();
|
||||
async move {
|
||||
// A failed exchange (malformed ping, timeout, etc.) is logged, not
|
||||
// surfaced as an accept error — one bad prober shouldn't disturb the
|
||||
// listener. Returning Ok keeps the router loop healthy.
|
||||
if let Err(e) = exchange(&connection, &handler).await {
|
||||
crate::log_msg(&format!("presence: inbound friends exchange failed: {e:#}"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
Reference in New Issue
Block a user