From c435881944d5d080a6f3f2298d7197654503bf3e Mon Sep 17 00:00:00 2001 From: Wind-Explorer Date: Mon, 31 Aug 2026 14:45:48 +0800 Subject: [PATCH] server priority --- src-common/src/lib.rs | 2 +- .../20260831000000_add_remote_priority.sql | 14 + src-tauri/src/lib.rs | 1 + src-tauri/src/live_data/mod.rs | 50 +++- src-tauri/src/network/connection.rs | 64 +++-- src-tauri/src/network/mod.rs | 263 +++++++++++++----- src-tauri/src/network/presence.rs | 64 ++++- src-tauri/src/network/routing.rs | 158 +++++++++++ src-tauri/src/remotes/mod.rs | 147 ++++++++-- src/lib/bindings.ts | 5 +- .../control-panel/network-panel.svelte | 67 ++++- 11 files changed, 681 insertions(+), 154 deletions(-) create mode 100644 src-tauri/migrations/20260831000000_add_remote_priority.sql create mode 100644 src-tauri/src/network/routing.rs diff --git a/src-common/src/lib.rs b/src-common/src/lib.rs index 6960734..c1e5976 100644 --- a/src-common/src/lib.rs +++ b/src-common/src/lib.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; use specta::Type; -pub const VERSION: u8 = 3; +pub const VERSION: u8 = 4; pub const MAX_INTERACTION_PAYLOAD_BYTES: usize = 160 * 1024; pub const MAX_IMAGE_B64_SIZE: usize = 150 * 1024; pub const MAX_SKIN_B64_SIZE: usize = 96 * 1024; diff --git a/src-tauri/migrations/20260831000000_add_remote_priority.sql b/src-tauri/migrations/20260831000000_add_remote_priority.sql new file mode 100644 index 0000000..d1e7d11 --- /dev/null +++ b/src-tauri/migrations/20260831000000_add_remote_priority.sql @@ -0,0 +1,14 @@ +ALTER TABLE remotes ADD COLUMN priority INTEGER NOT NULL DEFAULT 0; + +UPDATE remotes AS remote +SET priority = ( + SELECT COUNT(*) + FROM remotes AS preceding + WHERE COALESCE(preceding.name, preceding.address) COLLATE NOCASE + < COALESCE(remote.name, remote.address) COLLATE NOCASE + OR ( + COALESCE(preceding.name, preceding.address) COLLATE NOCASE + = COALESCE(remote.name, remote.address) COLLATE NOCASE + AND preceding.id < remote.id + ) +); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index d622f9a..adba484 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -73,6 +73,7 @@ fn specta_builder() -> tauri_specta::Builder { remotes::get_remote, remotes::update_remote, remotes::delete_remote, + remotes::reorder_remotes, profile::get_profile, profile::update_profile, profile::reset_profile_skin, diff --git a/src-tauri/src/live_data/mod.rs b/src-tauri/src/live_data/mod.rs index 609828f..a13ee50 100644 --- a/src-tauri/src/live_data/mod.rs +++ b/src-tauri/src/live_data/mod.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::hash::Hash; use serde::{Deserialize, Serialize}; use specta::Type; @@ -7,6 +8,27 @@ use tauri::{AppHandle, Manager}; use crate::cursor::CursorPositions; use crate::ufa::AppMeta; +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub(crate) enum LiveDataKind { + Cursor, + ForegroundApp, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LiveDataEnvelope { + pub session_id: String, + pub sequence: u64, + #[serde(flatten)] + pub data: LiveData, +} + +impl LiveDataEnvelope { + pub(crate) fn kind(&self) -> LiveDataKind { + self.data.kind() + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "camelCase")] pub enum LiveData { @@ -14,6 +36,15 @@ pub enum LiveData { ForegroundApp { meta: AppMeta }, } +impl LiveData { + pub(crate) fn kind(&self) -> LiveDataKind { + match self { + Self::Cursor { .. } => LiveDataKind::Cursor, + Self::ForegroundApp { .. } => LiveDataKind::ForegroundApp, + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, Type)] #[serde(rename_all = "camelCase")] pub struct LiveDataSnapshot { @@ -58,21 +89,28 @@ mod tests { #[test] fn live_data_is_internally_tagged_for_client_side_dispatch() { - let payload = serde_json::to_string(&LiveData::Cursor { - positions: CursorPositions { - raw: CursorPosition { x: 120.0, y: 80.0 }, - mapped: CursorPosition { x: 0.25, y: 0.5 }, + let payload = serde_json::to_string(&LiveDataEnvelope { + session_id: "session".to_owned(), + sequence: 7, + data: LiveData::Cursor { + positions: CursorPositions { + raw: CursorPosition { x: 120.0, y: 80.0 }, + mapped: CursorPosition { x: 0.25, y: 0.5 }, + }, }, }) .unwrap(); assert_eq!( payload, - r#"{"type":"cursor","positions":{"raw":{"x":120.0,"y":80.0},"mapped":{"x":0.25,"y":0.5}}}"# + r#"{"sessionId":"session","sequence":7,"type":"cursor","positions":{"raw":{"x":120.0,"y":80.0},"mapped":{"x":0.25,"y":0.5}}}"# ); assert!(matches!( serde_json::from_str(&payload).unwrap(), - LiveData::Cursor { .. } + LiveDataEnvelope { + data: LiveData::Cursor { .. }, + .. + } )); } } diff --git a/src-tauri/src/network/connection.rs b/src-tauri/src/network/connection.rs index a8b59ea..9220c43 100644 --- a/src-tauri/src/network/connection.rs +++ b/src-tauri/src/network/connection.rs @@ -18,7 +18,6 @@ use crate::db::AppDatabase; use crate::friends; use crate::interactions; use crate::keypair::AppKeypair; -use crate::live_data::LiveData; use crate::remotes::Remote; pub(super) struct InteractionRequest { @@ -45,7 +44,8 @@ pub(super) struct ConnectionInputs { pub(super) profiles: watch::Receiver, pub(super) friends: watch::Receiver>, pub(super) keypair: AppKeypair, - pub(super) live_data: mpsc::Receiver, + pub(super) cursor_data: watch::Receiver>, + pub(super) foreground_app_data: watch::Receiver>, pub(super) interactions: mpsc::Receiver, pub(super) profile_lookups: mpsc::Receiver, pub(super) skin_lookups: mpsc::Receiver, @@ -103,7 +103,8 @@ async fn connect( profiles, friends, keypair, - live_data, + cursor_data, + foreground_app_data, interactions: active_outgoing, profile_lookups, skin_lookups, @@ -141,7 +142,8 @@ async fn connect( if !matches!(recv(&mut reader).await?, ServerMessage::Registered) { return Err("server rejected registration".into()); } - while live_data.try_recv().is_ok() {} + cursor_data.borrow_and_update(); + foreground_app_data.borrow_and_update(); while let Ok(request) = active_outgoing.try_recv() { let _ = request .response @@ -166,12 +168,25 @@ async fn connect( > = HashMap::new(); loop { tokio::select! { - payload = live_data.recv() => { - let payload = payload.ok_or("network sender closed")?; - send(&mut writer, &ClientMessage::Signed { - signature: keypair.sign(&message_bytes(&payload)), - payload, - }).await?; + changed = cursor_data.changed() => { + changed.map_err(|_| "cursor sender closed")?; + let payload = { cursor_data.borrow_and_update().clone() }; + if let Some(payload) = payload { + send(&mut writer, &ClientMessage::Signed { + signature: keypair.sign(&message_bytes(&payload)), + payload, + }).await?; + } + } + changed = foreground_app_data.changed() => { + changed.map_err(|_| "foreground-app sender closed")?; + let payload = { foreground_app_data.borrow_and_update().clone() }; + if let Some(payload) = payload { + send(&mut writer, &ClientMessage::Signed { + signature: keypair.sign(&message_bytes(&payload)), + payload, + }).await?; + } } request = active_outgoing.recv() => { let request = request.ok_or("interaction sender closed")?; @@ -243,12 +258,19 @@ async fn connect( message = reader.next() => match message.ok_or("server closed the socket")?? { Message::Text(text) => match serde_json::from_str(&text)? { ServerMessage::FriendProfileUpdated { profile } => { - let database = handle.state::(); - if let Err(error) = friends::apply_profile_update(handle, &database, profile).await { - eprintln!("failed to update friend profile: {error}"); + if handle.state::().accept_source(&remote.id, &profile.id) { + let database = handle.state::(); + if let Err(error) = friends::apply_profile_update(handle, &database, profile).await { + eprintln!("failed to update friend profile: {error}"); + } } } ServerMessage::FriendProfiles { profiles } => { + let network = handle.state::(); + let profiles = profiles + .into_iter() + .filter(|profile| network.accept_source(&remote.id, &profile.id)) + .collect(); let database = handle.state::(); if let Err(error) = friends::apply_profile_sync(handle, &database, profiles).await { eprintln!("failed to synchronize friend profiles: {error}"); @@ -270,17 +292,11 @@ async fn connect( ); } ServerMessage::FriendLiveData { friend_id, payload } => { - match serde_json::from_str(&payload) { - Ok(LiveData::Cursor { positions }) => { - crate::cursor::emit_position(handle, friend_id, positions); - } - Ok(LiveData::ForegroundApp { meta }) => { - crate::ufa::emit_friend_app(handle, friend_id, meta); - } - Err(error) => { - eprintln!("failed to decode friend live data: {error}"); - } - } + handle.state::().receive_live_data( + handle, + friend_id, + &payload, + ); } ServerMessage::FriendInteraction { interaction_id, diff --git a/src-tauri/src/network/mod.rs b/src-tauri/src/network/mod.rs index c202ebe..9c91317 100644 --- a/src-tauri/src/network/mod.rs +++ b/src-tauri/src/network/mod.rs @@ -3,8 +3,6 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use futures_util::StreamExt; -use futures_util::stream::FuturesUnordered; use serde::{Deserialize, Serialize}; use specta::Type; use tauri::{AppHandle, Manager, State}; @@ -15,18 +13,20 @@ use wyd_common::{InteractionContent, InteractionDeliveryStatus}; use crate::db::AppDatabase; use crate::friends::{self, FriendsChanged}; use crate::keypair::AppKeypair; -use crate::live_data::LiveData; +use crate::live_data::{LiveData, LiveDataEnvelope, LiveDataKind}; use crate::remotes::{self, Remote, RemotesChanged}; mod connection; mod presence; +mod routing; use connection::{ConnectionInputs, InteractionRequest, ProfileLookupRequest, SkinLookupRequest}; use presence::{Change as FriendPresenceChange, FriendPresence}; +use routing::SequenceTracker; type Statuses = Arc>>; -#[derive(Debug, Clone, Serialize, Deserialize, Type)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Type)] #[serde(rename_all = "camelCase")] pub enum ConnectionState { Connecting, @@ -57,7 +57,8 @@ pub struct FriendStatusesChanged { struct Connection { remote: Remote, - sender: mpsc::Sender, + cursor_sender: watch::Sender>, + foreground_app_sender: watch::Sender>, interaction_sender: mpsc::Sender, profile_lookup_sender: mpsc::Sender, skin_lookup_sender: mpsc::Sender, @@ -65,6 +66,15 @@ struct Connection { generation: u64, } +#[derive(Clone)] +struct ConnectionSenders { + remote_id: String, + priority: i32, + interaction: mpsc::Sender, + profile_lookup: mpsc::Sender, + skin_lookup: mpsc::Sender, +} + pub struct Network { connections: Mutex>, statuses: Statuses, @@ -73,6 +83,10 @@ pub struct Network { friends: watch::Sender>, keypair: AppKeypair, next_generation: AtomicU64, + live_session_id: String, + next_cursor_sequence: AtomicU64, + next_foreground_app_sequence: AtomicU64, + received_sequences: SequenceTracker, } impl Network { @@ -81,7 +95,18 @@ impl Network { } pub fn send_live_data(&self, data: LiveData) { - let Ok(payload) = serde_json::to_string(&data) else { + let sequence = match data.kind() { + LiveDataKind::Cursor => self.next_cursor_sequence.fetch_add(1, Ordering::Relaxed), + LiveDataKind::ForegroundApp => self + .next_foreground_app_sequence + .fetch_add(1, Ordering::Relaxed), + }; + let envelope = LiveDataEnvelope { + session_id: self.live_session_id.clone(), + sequence, + data, + }; + let Ok(payload) = serde_json::to_string(&envelope) else { eprintln!("failed to serialize live data"); return; }; @@ -90,11 +115,63 @@ impl Network { return; }; for connection in connections.values() { - match connection.sender.try_send(payload.clone()) { - Ok(()) | Err(mpsc::error::TrySendError::Full(_)) => {} - Err(mpsc::error::TrySendError::Closed(_)) => { - eprintln!("remote {} cannot accept live data", connection.remote.id); + match envelope.kind() { + LiveDataKind::Cursor => { + connection.cursor_sender.send_replace(Some(payload.clone())); } + LiveDataKind::ForegroundApp => { + connection + .foreground_app_sender + .send_replace(Some(payload.clone())); + } + } + } + } + + fn preferred_remote( + &self, + friend_id: &str, + incoming_remote_id: Option<&str>, + ) -> Option { + let mut remote_ids = self.friend_presence.remotes_for(friend_id).ok()?; + if let Some(remote_id) = incoming_remote_id { + remote_ids.insert(remote_id.to_owned()); + } + let connections = self.connections.lock().ok()?; + let priorities = connections + .iter() + .map(|(remote_id, connection)| (remote_id.clone(), connection.remote.priority)) + .collect(); + routing::preferred_remote(&remote_ids, &priorities) + } + + pub(super) fn accept_source(&self, remote_id: &str, friend_id: &str) -> bool { + self.preferred_remote(friend_id, Some(remote_id)) + .is_some_and(|preferred| preferred == remote_id) + } + + fn receive_live_data(&self, handle: &AppHandle, friend_id: String, payload: &str) { + let envelope = match serde_json::from_str::(payload) { + Ok(envelope) => envelope, + Err(error) => { + eprintln!("failed to decode friend live data: {error}"); + return; + } + }; + match self.received_sequences.accept(&friend_id, &envelope) { + Ok(true) => {} + Ok(false) => return, + Err(error) => { + eprintln!("failed to track friend live data: {error}"); + return; + } + } + match envelope.data { + LiveData::Cursor { positions } => { + crate::cursor::emit_position(handle, friend_id, positions); + } + LiveData::ForegroundApp { meta } => { + crate::ufa::emit_friend_app(handle, friend_id, meta); } } } @@ -103,6 +180,35 @@ impl Network { self.profile.send_replace(profile); } + fn ordered_senders(&self, friend_id: Option<&str>) -> Result, String> { + let routes = friend_id + .map(|friend_id| self.friend_presence.remotes_for(friend_id)) + .transpose()? + .unwrap_or_default(); + let connections = self.connections.lock().map_err(|error| error.to_string())?; + let statuses = self.statuses.lock().map_err(|error| error.to_string())?; + let mut senders = connections + .values() + .filter(|connection| { + statuses + .get(&connection.remote.id) + .is_some_and(|(_, status)| status.state == ConnectionState::Connected) + }) + .filter(|connection| routes.is_empty() || routes.contains(&connection.remote.id)) + .map(|connection| ConnectionSenders { + remote_id: connection.remote.id.clone(), + priority: connection.remote.priority, + interaction: connection.interaction_sender.clone(), + profile_lookup: connection.profile_lookup_sender.clone(), + skin_lookup: connection.skin_lookup_sender.clone(), + }) + .collect::>(); + senders.sort_by(|left, right| { + (left.priority, &left.remote_id).cmp(&(right.priority, &right.remote_id)) + }); + Ok(senders) + } + pub async fn send_interaction( &self, recipient_id: String, @@ -113,18 +219,13 @@ impl Network { } let payload = serde_json::to_string(&content).map_err(|error| error.to_string())?; let interaction_id = uuid::Uuid::new_v4().to_string(); - let senders: Vec<_> = self - .connections - .lock() - .map_err(|error| error.to_string())? - .values() - .map(|connection| connection.interaction_sender.clone()) - .collect(); + let senders = self.ordered_senders(Some(&recipient_id))?; if senders.is_empty() { - return Err("No relay connections are configured".to_owned()); + return Err("No connected relay can currently reach this friend".to_owned()); } - let mut responses = Vec::new(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + let mut statuses = Vec::new(); for sender in senders { let (response, receiver) = oneshot::channel(); let request = InteractionRequest { @@ -133,20 +234,15 @@ impl Network { payload: payload.clone(), response, }; - if sender.try_send(request).is_ok() { - responses.push(receiver); + if sender.interaction.try_send(request).is_err() { + continue; } - } - if responses.is_empty() { - return Err("Relay connections are busy or disconnected".to_owned()); - } - - let mut pending: FuturesUnordered<_> = responses - .into_iter() - .map(|response| tokio::time::timeout(Duration::from_secs(5), response)) - .collect(); - let mut statuses = Vec::new(); - while let Some(result) = pending.next().await { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + break; + } + let result = + tokio::time::timeout(remaining.min(Duration::from_secs(2)), receiver).await; if let Ok(Ok(status)) = result { if status == InteractionDeliveryStatus::Delivered { return Ok(()); @@ -169,15 +265,9 @@ impl Network { return Err("You cannot add your own identification key.".to_owned()); } - let senders: Vec<_> = self - .connections - .lock() - .map_err(|error| error.to_string())? - .values() - .map(|connection| connection.profile_lookup_sender.clone()) - .collect(); + let senders = self.ordered_senders(None)?; let request_id = uuid::Uuid::new_v4().to_string(); - let mut responses = Vec::new(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(3); for sender in senders { let (response, receiver) = oneshot::channel(); let request = ProfileLookupRequest { @@ -185,16 +275,15 @@ impl Network { user_id: user_id.clone(), response, }; - if sender.try_send(request).is_ok() { - responses.push(receiver); + if sender.profile_lookup.try_send(request).is_err() { + continue; } - } - - let mut pending: FuturesUnordered<_> = responses - .into_iter() - .map(|response| tokio::time::timeout(Duration::from_secs(3), response)) - .collect(); - while let Some(result) = pending.next().await { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + break; + } + let result = + tokio::time::timeout(remaining.min(Duration::from_secs(1)), receiver).await; if let Ok(Ok(Some(display_name))) = result { return Ok(Some(display_name)); } @@ -211,35 +300,29 @@ impl Network { return Ok(None); } - let senders: Vec<_> = self - .connections - .lock() - .map_err(|error| error.to_string())? - .values() - .map(|connection| connection.skin_lookup_sender.clone()) - .collect(); + let senders = self.ordered_senders(Some(&user_id))?; let request_id = uuid::Uuid::new_v4().to_string(); - let mut responses = Vec::new(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(3); for sender in senders { let (response, receiver) = oneshot::channel(); if sender + .skin_lookup .try_send(SkinLookupRequest { request_id: request_id.clone(), user_id: user_id.clone(), skin_hash: skin_hash.clone(), response, }) - .is_ok() + .is_err() { - responses.push(receiver); + continue; } - } - - let mut pending: FuturesUnordered<_> = responses - .into_iter() - .map(|response| tokio::time::timeout(Duration::from_secs(3), response)) - .collect(); - while let Some(result) = pending.next().await { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + break; + } + let result = + tokio::time::timeout(remaining.min(Duration::from_secs(1)), receiver).await; if let Ok(Ok(Some(data))) = result && let Some(bytes) = crate::skins::decode_response(&data, &skin_hash) { @@ -258,7 +341,11 @@ impl Network { let stale: Vec<_> = connections .iter() - .filter(|(id, connection)| desired.get(*id) != Some(&connection.remote)) + .filter(|(id, connection)| { + desired + .get(*id) + .is_none_or(|remote| !same_connection_configuration(remote, &connection.remote)) + }) .map(|(id, _)| id.clone()) .collect(); let mut presence_changes = Vec::new(); @@ -271,13 +358,17 @@ impl Network { } } + let mut priority_changed = false; for remote in desired.into_values() { - if connections.contains_key(&remote.id) { + if let Some(connection) = connections.get_mut(&remote.id) { + priority_changed |= connection.remote.priority != remote.priority; + connection.remote = remote; continue; } let generation = self.next_generation.fetch_add(1, Ordering::Relaxed); - let (sender, receiver) = mpsc::channel(32); + let (cursor_sender, cursor_receiver) = watch::channel(None); + let (foreground_app_sender, foreground_app_receiver) = watch::channel(None); let (interaction_sender, interaction_receiver) = mpsc::channel(16); let (profile_lookup_sender, profile_lookup_receiver) = mpsc::channel(16); let (skin_lookup_sender, skin_lookup_receiver) = mpsc::channel(16); @@ -297,7 +388,8 @@ impl Network { profiles: self.profile.subscribe(), friends: self.friends.subscribe(), keypair: self.keypair.clone(), - live_data: receiver, + cursor_data: cursor_receiver, + foreground_app_data: foreground_app_receiver, interactions: interaction_receiver, profile_lookups: profile_lookup_receiver, skin_lookups: skin_lookup_receiver, @@ -307,7 +399,8 @@ impl Network { remote.id.clone(), Connection { remote, - sender, + cursor_sender, + foreground_app_sender, interaction_sender, profile_lookup_sender, skin_lookup_sender, @@ -321,6 +414,9 @@ impl Network { for change in presence_changes { apply_friend_presence_change(handle, change); } + if priority_changed && let Err(error) = crate::live_data::publish_current(handle) { + eprintln!("failed to republish live data after server reorder: {error}"); + } emit_statuses(handle, &self.statuses) } } @@ -340,6 +436,10 @@ pub async fn init(handle: &AppHandle) -> Result<(), Box> friends, keypair, next_generation: AtomicU64::new(1), + live_session_id: uuid::Uuid::new_v4().to_string(), + next_cursor_sequence: AtomicU64::new(1), + next_foreground_app_sequence: AtomicU64::new(1), + received_sequences: SequenceTracker::default(), }; handle.manage(network); handle @@ -365,6 +465,9 @@ pub async fn init(handle: &AppHandle) -> Result<(), Box> *current = ids.clone(); true }); + if let Err(error) = network.received_sequences.retain(&ids) { + eprintln!("failed to prune live-data sequences: {error}"); + } apply_friend_presence_change(&listener_handle, network.friend_presence.retain(&ids)); }); Ok(()) @@ -423,15 +526,16 @@ fn apply_friend_presence_change( { eprintln!("failed to remove offline foreground apps: {error}"); } - if !change.came_online.is_empty() + if change.route_added && let Err(error) = crate::live_data::publish_current(handle) { eprintln!("failed to publish current live data: {error}"); } - if let Err(error) = (FriendStatusesChanged { - friend_ids: change.online, - }) - .emit(handle) + if change.online_changed + && let Err(error) = (FriendStatusesChanged { + friend_ids: change.online, + }) + .emit(handle) { eprintln!("failed to emit friend statuses: {error}"); } @@ -522,6 +626,13 @@ fn friend_ids(friends: Vec, own_id: &str) -> Vec ids } +fn same_connection_configuration(left: &Remote, right: &Remote) -> bool { + left.id == right.id + && left.address == right.address + && left.name == right.name + && left.port == right.port +} + #[cfg(test)] mod tests { use super::friend_ids; diff --git a/src-tauri/src/network/presence.rs b/src-tauri/src/network/presence.rs index 7e5a2f3..be2d13f 100644 --- a/src-tauri/src/network/presence.rs +++ b/src-tauri/src/network/presence.rs @@ -6,19 +6,30 @@ pub(super) struct Change { pub(super) online: Vec, pub(super) came_online: Vec, pub(super) went_offline: Vec, + pub(super) online_changed: bool, + pub(super) route_added: bool, } #[derive(Default)] pub(super) struct FriendPresence(Mutex>>); impl FriendPresence { + pub(super) fn remotes_for(&self, friend_id: &str) -> Result, String> { + let by_remote = self.0.lock().map_err(|error| error.to_string())?; + Ok(by_remote + .iter() + .filter(|(_, friend_ids)| friend_ids.contains(friend_id)) + .map(|(remote_id, _)| remote_id.clone()) + .collect()) + } + pub(super) fn replace( &self, remote_id: &str, friend_ids: Vec, ) -> Result, String> { let mut by_remote = self.0.lock().map_err(|error| error.to_string())?; - let before = aggregate(&by_remote); + let before = by_remote.clone(); let friend_ids = friend_ids.into_iter().collect::>(); if friend_ids.is_empty() { by_remote.remove(remote_id); @@ -35,7 +46,7 @@ impl FriendPresence { online: bool, ) -> Result, String> { let mut by_remote = self.0.lock().map_err(|error| error.to_string())?; - let before = aggregate(&by_remote); + let before = by_remote.clone(); if online { by_remote .entry(remote_id.to_owned()) @@ -52,7 +63,7 @@ impl FriendPresence { pub(super) fn remove(&self, remote_id: &str) -> Result, String> { let mut by_remote = self.0.lock().map_err(|error| error.to_string())?; - let before = aggregate(&by_remote); + let before = by_remote.clone(); by_remote.remove(remote_id); Ok(diff(before, &by_remote)) } @@ -60,7 +71,7 @@ impl FriendPresence { pub(super) fn retain(&self, known_friend_ids: &[String]) -> Result, String> { let known_friend_ids = known_friend_ids.iter().collect::>(); let mut by_remote = self.0.lock().map_err(|error| error.to_string())?; - let before = aggregate(&by_remote); + let before = by_remote.clone(); by_remote.retain(|_, friend_ids| { friend_ids.retain(|friend_id| known_friend_ids.contains(friend_id)); !friend_ids.is_empty() @@ -83,11 +94,21 @@ fn aggregate(by_remote: &HashMap>) -> HashSet { .collect() } -fn diff(before: HashSet, by_remote: &HashMap>) -> Option { - let after = aggregate(by_remote); - if before == after { +fn diff( + before_by_remote: HashMap>, + by_remote: &HashMap>, +) -> Option { + if before_by_remote == *by_remote { return None; } + let before = aggregate(&before_by_remote); + let after = aggregate(by_remote); + let online_changed = before != after; + let route_added = by_remote.iter().any(|(remote_id, friend_ids)| { + before_by_remote + .get(remote_id) + .is_none_or(|before| !friend_ids.is_subset(before)) + }); let mut went_offline = before.difference(&after).cloned().collect::>(); let mut came_online = after.difference(&before).cloned().collect::>(); let mut online = after.into_iter().collect::>(); @@ -98,6 +119,8 @@ fn diff(before: HashSet, by_remote: &HashMap>) - online, came_online, went_offline, + online_changed, + route_added, }) } @@ -116,21 +139,40 @@ mod tests { online: vec!["friend".to_owned()], came_online: vec!["friend".to_owned()], went_offline: Vec::new(), + online_changed: true, + route_added: true, }) ); - assert!( + assert_eq!( presence .replace("remote-b", vec!["friend".to_owned()]) - .unwrap() - .is_none() + .unwrap(), + Some(Change { + online: vec!["friend".to_owned()], + came_online: Vec::new(), + went_offline: Vec::new(), + online_changed: false, + route_added: true, + }) + ); + assert_eq!( + presence.remove("remote-a").unwrap(), + Some(Change { + online: vec!["friend".to_owned()], + came_online: Vec::new(), + went_offline: Vec::new(), + online_changed: false, + route_added: false, + }) ); - assert!(presence.remove("remote-a").unwrap().is_none()); assert_eq!( presence.remove("remote-b").unwrap(), Some(Change { online: Vec::new(), came_online: Vec::new(), went_offline: vec!["friend".to_owned()], + online_changed: true, + route_added: false, }) ); } diff --git a/src-tauri/src/network/routing.rs b/src-tauri/src/network/routing.rs new file mode 100644 index 0000000..33988a7 --- /dev/null +++ b/src-tauri/src/network/routing.rs @@ -0,0 +1,158 @@ +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::Mutex; + +use crate::live_data::{LiveDataEnvelope, LiveDataKind}; + +const RETIRED_SESSION_LIMIT: usize = 4; + +pub(super) fn preferred_remote( + remote_ids: &HashSet, + priorities: &HashMap, +) -> Option { + remote_ids + .iter() + .filter_map(|remote_id| { + priorities + .get(remote_id) + .map(|priority| (*priority, remote_id)) + }) + .min() + .map(|(_, remote_id)| remote_id.clone()) +} + +#[derive(Debug, Hash, PartialEq, Eq)] +struct StreamKey { + friend_id: String, + kind: LiveDataKind, +} + +#[derive(Debug)] +struct StreamSequence { + session_id: String, + sequence: u64, + retired_sessions: VecDeque, +} + +#[derive(Default)] +pub(super) struct SequenceTracker(Mutex>); + +impl SequenceTracker { + pub(super) fn accept( + &self, + friend_id: &str, + envelope: &LiveDataEnvelope, + ) -> Result { + if envelope.sequence == 0 || uuid::Uuid::parse_str(&envelope.session_id).is_err() { + return Ok(false); + } + + let key = StreamKey { + friend_id: friend_id.to_owned(), + kind: envelope.kind(), + }; + let mut streams = self.0.lock().map_err(|error| error.to_string())?; + let Some(current) = streams.get_mut(&key) else { + streams.insert( + key, + StreamSequence { + session_id: envelope.session_id.clone(), + sequence: envelope.sequence, + retired_sessions: VecDeque::new(), + }, + ); + return Ok(true); + }; + + if current.session_id == envelope.session_id { + if envelope.sequence <= current.sequence { + return Ok(false); + } + current.sequence = envelope.sequence; + return Ok(true); + } + if current.retired_sessions.contains(&envelope.session_id) { + return Ok(false); + } + + current + .retired_sessions + .push_back(current.session_id.clone()); + if current.retired_sessions.len() > RETIRED_SESSION_LIMIT { + current.retired_sessions.pop_front(); + } + current.session_id = envelope.session_id.clone(); + current.sequence = envelope.sequence; + Ok(true) + } + + pub(super) fn retain(&self, friend_ids: &[String]) -> Result<(), String> { + let mut streams = self.0.lock().map_err(|error| error.to_string())?; + streams.retain(|key, _| friend_ids.contains(&key.friend_id)); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::collections::{HashMap, HashSet}; + + use super::{SequenceTracker, preferred_remote}; + use crate::cursor::{CursorPosition, CursorPositions}; + use crate::live_data::{LiveData, LiveDataEnvelope}; + + fn envelope(session_id: &str, sequence: u64) -> LiveDataEnvelope { + LiveDataEnvelope { + session_id: session_id.to_owned(), + sequence, + data: LiveData::Cursor { + positions: CursorPositions { + raw: CursorPosition::default(), + mapped: CursorPosition::default(), + }, + }, + } + } + + #[test] + fn rejects_duplicate_out_of_order_and_retired_session_packets() { + let tracker = SequenceTracker::default(); + let first = uuid::Uuid::new_v4().to_string(); + let second = uuid::Uuid::new_v4().to_string(); + + assert!(tracker.accept("friend", &envelope(&first, 2)).unwrap()); + assert!(!tracker.accept("friend", &envelope(&first, 2)).unwrap()); + assert!(!tracker.accept("friend", &envelope(&first, 1)).unwrap()); + assert!(tracker.accept("friend", &envelope(&second, 1)).unwrap()); + assert!(!tracker.accept("friend", &envelope(&first, 3)).unwrap()); + assert!(tracker.accept("friend", &envelope(&second, 2)).unwrap()); + } + + #[test] + fn relay_duplicates_do_not_block_a_newer_stream_packet() { + let tracker = SequenceTracker::default(); + let session = uuid::Uuid::new_v4().to_string(); + + assert!(tracker.accept("friend", &envelope(&session, 7)).unwrap()); + assert!(!tracker.accept("friend", &envelope(&session, 7)).unwrap()); + assert!(tracker.accept("friend", &envelope(&session, 8)).unwrap()); + } + + #[test] + fn preferred_remote_uses_priority_then_stable_id() { + let sources = HashSet::from([ + "lower".to_owned(), + "higher-b".to_owned(), + "higher-a".to_owned(), + ]); + let priorities = HashMap::from([ + ("lower".to_owned(), 3), + ("higher-b".to_owned(), 1), + ("higher-a".to_owned(), 1), + ]); + + assert_eq!( + preferred_remote(&sources, &priorities).as_deref(), + Some("higher-a") + ); + } +} diff --git a/src-tauri/src/remotes/mod.rs b/src-tauri/src/remotes/mod.rs index 615bb3e..87cb579 100644 --- a/src-tauri/src/remotes/mod.rs +++ b/src-tauri/src/remotes/mod.rs @@ -12,6 +12,7 @@ pub struct Remote { pub address: String, pub name: Option, pub port: Option, + pub priority: i32, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Type)] @@ -30,36 +31,43 @@ pub struct RemotesChanged { pub async fn all(database: &AppDatabase) -> Result, sqlx::Error> { sqlx::query_as::<_, Remote>( - "SELECT id, address, name, port FROM remotes \ - ORDER BY COALESCE(name, address) COLLATE NOCASE, id", + "SELECT id, address, name, port, priority FROM remotes \ + ORDER BY priority, id", ) .fetch_all(database.pool()) .await } async fn get(database: &AppDatabase, id: &str) -> Result, sqlx::Error> { - sqlx::query_as::<_, Remote>("SELECT id, address, name, port FROM remotes WHERE id = ?1") - .bind(id) - .fetch_optional(database.pool()) - .await + sqlx::query_as::<_, Remote>( + "SELECT id, address, name, port, priority FROM remotes WHERE id = ?1", + ) + .bind(id) + .fetch_optional(database.pool()) + .await } async fn insert(database: &AppDatabase, input: RemoteInput) -> Result { + let id = Uuid::new_v4().to_string(); + let priority = sqlx::query_scalar::<_, i32>( + "INSERT INTO remotes (id, address, name, port, priority) \ + SELECT ?1, ?2, ?3, ?4, COALESCE(MAX(priority) + 1, 0) FROM remotes \ + RETURNING priority", + ) + .bind(&id) + .bind(&input.address) + .bind(&input.name) + .bind(input.port) + .fetch_one(database.pool()) + .await?; let remote = Remote { - id: Uuid::new_v4().to_string(), + id, address: input.address, name: input.name, port: input.port, + priority, }; - sqlx::query("INSERT INTO remotes (id, address, name, port) VALUES (?1, ?2, ?3, ?4)") - .bind(&remote.id) - .bind(&remote.address) - .bind(&remote.name) - .bind(remote.port) - .execute(database.pool()) - .await?; - Ok(remote) } @@ -80,12 +88,41 @@ async fn update( return Ok(None); } - Ok(Some(Remote { - id: id.to_owned(), - address: input.address, - name: input.name, - port: input.port, - })) + get(database, id).await +} + +async fn reorder(database: &AppDatabase, ids: &[String]) -> Result { + let existing = all(database).await.map_err(db::command_error)?; + if ids.len() != existing.len() { + return Err("Server order must contain every configured server exactly once".to_owned()); + } + + let mut expected = existing + .into_iter() + .map(|remote| remote.id) + .collect::>(); + let mut provided = ids.to_vec(); + expected.sort_unstable(); + provided.sort_unstable(); + provided.dedup(); + if provided != expected { + return Err("Server order contains an unknown or duplicate server".to_owned()); + } + + let mut transaction = database.pool().begin().await.map_err(db::command_error)?; + let mut changed = false; + for (priority, id) in ids.iter().enumerate() { + let result = + sqlx::query("UPDATE remotes SET priority = ?1 WHERE id = ?2 AND priority != ?1") + .bind(priority as i32) + .bind(id) + .execute(&mut *transaction) + .await + .map_err(db::command_error)?; + changed |= result.rows_affected() > 0; + } + transaction.commit().await.map_err(db::command_error)?; + Ok(changed) } async fn delete(database: &AppDatabase, id: &str) -> Result { @@ -178,6 +215,19 @@ pub async fn delete_remote( Ok(changed) } +#[tauri::command] +#[specta::specta] +pub async fn reorder_remotes( + handle: AppHandle, + database: State<'_, AppDatabase>, + ids: Vec, +) -> Result, String> { + if reorder(&database, &ids).await? { + emit_changed(&handle, &database).await?; + } + all(&database).await.map_err(db::command_error) +} + #[cfg(test)] mod tests { use sqlx::sqlite::SqlitePoolOptions; @@ -216,6 +266,7 @@ mod tests { assert_eq!(remote.address, "play.example.com"); assert_eq!(remote.name, None); assert_eq!(remote.port, None); + assert_eq!(remote.priority, 0); assert_eq!(get(&database, &remote.id).await.unwrap(), Some(remote)); } @@ -250,6 +301,7 @@ mod tests { assert_eq!(updated.address, "new.example.com"); assert_eq!(updated.name.as_deref(), Some("New name")); assert_eq!(updated.port, None); + assert_eq!(updated.priority, created.priority); assert_eq!(all(&database).await.unwrap(), vec![updated.clone()]); assert!(delete(&database, &updated.id).await.unwrap()); assert!(!delete(&database, &updated.id).await.unwrap()); @@ -274,4 +326,57 @@ mod tests { assert_eq!(updated, None); } + + #[tokio::test] + async fn reorder_validates_and_persists_the_complete_priority_order() { + let database = database().await; + let first = insert( + &database, + RemoteInput { + address: "first.example.com".to_owned(), + name: None, + port: None, + }, + ) + .await + .unwrap(); + let second = insert( + &database, + RemoteInput { + address: "second.example.com".to_owned(), + name: None, + port: None, + }, + ) + .await + .unwrap(); + + assert!( + reorder(&database, &[second.id.clone(), first.id.clone()]) + .await + .unwrap() + ); + let ordered = all(&database).await.unwrap(); + assert_eq!( + ordered.iter().map(|remote| &remote.id).collect::>(), + [&second.id, &first.id] + ); + assert_eq!( + ordered + .iter() + .map(|remote| remote.priority) + .collect::>(), + [0, 1] + ); + assert!( + !reorder(&database, &[second.id.clone(), first.id.clone()]) + .await + .unwrap() + ); + assert!( + reorder(&database, std::slice::from_ref(&first.id)) + .await + .is_err() + ); + } } diff --git a/src/lib/bindings.ts b/src/lib/bindings.ts index 09b82bf..be316ed 100644 --- a/src/lib/bindings.ts +++ b/src/lib/bindings.ts @@ -32,6 +32,9 @@ async updateRemote(id: string, remote: RemoteInput) : Promise { async deleteRemote(id: string) : Promise { return await TAURI_INVOKE("delete_remote", { id }); }, +async reorderRemotes(ids: string[]) : Promise { + return await TAURI_INVOKE("reorder_remotes", { ids }); +}, async getProfile() : Promise { return await TAURI_INVOKE("get_profile"); }, @@ -154,7 +157,7 @@ export type ProfileChanged = { profile: User } export type PuppetMovementMode = "free" | "bottom" export type PuppetState = { id: string; position: CursorPosition; isMoving: boolean } export type PuppetStatesChanged = { puppets: PuppetState[] } -export type Remote = { id: string; address: string; name: string | null; port: number | null } +export type Remote = { id: string; address: string; name: string | null; port: number | null; priority: number } export type RemoteInput = { address: string; name: string | null; port: number | null } export type RemotesChanged = { remotes: Remote[] } export type SceneConfiguration = { puppetScale: number; puppetOpacity: number; puppetMovementMode: PuppetMovementMode } diff --git a/src/lib/components/control-panel/network-panel.svelte b/src/lib/components/control-panel/network-panel.svelte index 03e7975..6a1cace 100644 --- a/src/lib/components/control-panel/network-panel.svelte +++ b/src/lib/components/control-panel/network-panel.svelte @@ -67,6 +67,24 @@ } } + async function moveRemote(id: string, offset: -1 | 1) { + const index = $remotes.findIndex((remote) => remote.id === id); + const target = index + offset; + if (index < 0 || target < 0 || target >= $remotes.length) return; + + const ordered = [...$remotes]; + [ordered[index], ordered[target]] = [ordered[target], ordered[index]]; + busy = true; + error = ""; + try { + await commands.reorderRemotes(ordered.map((remote) => remote.id)); + } catch (cause) { + error = String(cause); + } finally { + busy = false; + } + } + function reset() { mode = "browse"; error = ""; @@ -119,13 +137,13 @@ class="flex min-h-0 flex-1 flex-col border border-base-300 bg-base-100" >
- ServerConnection + ServerConnectionPriority
{#if $remotes.length === 0} @@ -138,17 +156,20 @@

{:else} - {#each $remotes as remote (remote.id)} + {#each $remotes as remote, index (remote.id)} {@const state = statusFor(remote.id)} - {state} - + + + + + {/each} {/if}