wind-explorer/friendolls
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

server priority

Browse Source
This commit is contained in:
wind-explorer
2026-08-31 14:45:48 +08:00
parent f82d1041ae
commit c435881944
11 changed files with 681 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-common/src/lib.rs
+1 -1
View File
@@ -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;
src-tauri/migrations/20260831000000_add_remote_priority.sql
+14
View File
@@ -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
)
);
src-tauri/src/lib.rs
+1
View File
@@ -73,6 +73,7 @@ fn specta_builder() -> tauri_specta::Builder<tauri::Wry> {
remotes::get_remote,
remotes::update_remote,
remotes::delete_remote,
remotes::reorder_remotes,
profile::get_profile,
profile::update_profile,
profile::reset_profile_skin,
src-tauri/src/live_data/mod.rs
+44 -6
View File
@@ -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 { .. },
..
}
));
}
}
src-tauri/src/network/connection.rs
+40 -24
View File
@@ -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<crate::user::User>,
pub(super) friends: watch::Receiver<Vec<String>>,
pub(super) keypair: AppKeypair,
pub(super) live_data: mpsc::Receiver<String>,
pub(super) cursor_data: watch::Receiver<Option<String>>,
pub(super) foreground_app_data: watch::Receiver<Option<String>>,
pub(super) interactions: mpsc::Receiver<InteractionRequest>,
pub(super) profile_lookups: mpsc::Receiver<ProfileLookupRequest>,
pub(super) skin_lookups: mpsc::Receiver<SkinLookupRequest>,
@@ -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::<AppDatabase>();
if let Err(error) = friends::apply_profile_update(handle, &database, profile).await {
eprintln!("failed to update friend profile: {error}");
if handle.state::<super::Network>().accept_source(&remote.id, &profile.id) {
let database = handle.state::<AppDatabase>();
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::<super::Network>();
let profiles = profiles
.into_iter()
.filter(|profile| network.accept_source(&remote.id, &profile.id))
.collect();
let database = handle.state::<AppDatabase>();
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::<super::Network>().receive_live_data(
handle,
friend_id,
&payload,
);
}
ServerMessage::FriendInteraction {
interaction_id,
src-tauri/src/network/mod.rs
+187 -76
View File
@@ -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<Mutex<HashMap<String, (u64, ConnectionStatus)>>>;
#[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<String>,
cursor_sender: watch::Sender<Option<String>>,
foreground_app_sender: watch::Sender<Option<String>>,
interaction_sender: mpsc::Sender<InteractionRequest>,
profile_lookup_sender: mpsc::Sender<ProfileLookupRequest>,
skin_lookup_sender: mpsc::Sender<SkinLookupRequest>,
@@ -65,6 +66,15 @@ struct Connection {
generation: u64,
}
#[derive(Clone)]
struct ConnectionSenders {
remote_id: String,
priority: i32,
interaction: mpsc::Sender<InteractionRequest>,
profile_lookup: mpsc::Sender<ProfileLookupRequest>,
skin_lookup: mpsc::Sender<SkinLookupRequest>,
}
pub struct Network {
connections: Mutex<HashMap<String, Connection>>,
statuses: Statuses,
@@ -73,6 +83,10 @@ pub struct Network {
friends: watch::Sender<Vec<String>>,
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<String> {
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::<LiveDataEnvelope>(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<Vec<ConnectionSenders>, 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::<Vec<_>>();
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<dyn std::error::Error>>
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<dyn std::error::Error>>
*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<crate::friends::Friend>, own_id: &str) -> Vec<String>
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;
src-tauri/src/network/presence.rs
+53 -11
View File
@@ -6,19 +6,30 @@ pub(super) struct Change {
pub(super) online: Vec<String>,
pub(super) came_online: Vec<String>,
pub(super) went_offline: Vec<String>,
pub(super) online_changed: bool,
pub(super) route_added: bool,
}
#[derive(Default)]
pub(super) struct FriendPresence(Mutex<HashMap<String, HashSet<String>>>);
impl FriendPresence {
pub(super) fn remotes_for(&self, friend_id: &str) -> Result<HashSet<String>, 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<String>,
) -> Result<Option<Change>, 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::<HashSet<_>>();
if friend_ids.is_empty() {
by_remote.remove(remote_id);
@@ -35,7 +46,7 @@ impl FriendPresence {
online: bool,
) -> Result<Option<Change>, 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<Option<Change>, 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<Option<Change>, String> {
let known_friend_ids = known_friend_ids.iter().collect::<HashSet<_>>();
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<String, HashSet<String>>) -> HashSet<String> {
.collect()
}
fn diff(before: HashSet<String>, by_remote: &HashMap<String, HashSet<String>>) -> Option<Change> {
let after = aggregate(by_remote);
if before == after {
fn diff(
before_by_remote: HashMap<String, HashSet<String>>,
by_remote: &HashMap<String, HashSet<String>>,
) -> Option<Change> {
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::<Vec<_>>();
let mut came_online = after.difference(&before).cloned().collect::<Vec<_>>();
let mut online = after.into_iter().collect::<Vec<_>>();
@@ -98,6 +119,8 @@ fn diff(before: HashSet<String>, by_remote: &HashMap<String, HashSet<String>>) -
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,
})
);
}
src-tauri/src/network/routing.rs
+158
View File
@@ -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<String>,
priorities: &HashMap<String, i32>,
) -> Option<String> {
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<String>,
}
#[derive(Default)]
pub(super) struct SequenceTracker(Mutex<HashMap<StreamKey, StreamSequence>>);
impl SequenceTracker {
pub(super) fn accept(
&self,
friend_id: &str,
envelope: &LiveDataEnvelope,
) -> Result<bool, String> {
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")
);
}
}
src-tauri/src/remotes/mod.rs
+126 -21
View File
@@ -12,6 +12,7 @@ pub struct Remote {
pub address: String,
pub name: Option<String>,
pub port: Option<u16>,
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<Vec<Remote>, 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<Option<Remote>, 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<Remote, sqlx::Error> {
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<bool, String> {
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::<Vec<_>>();
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<bool, sqlx::Error> {
@@ -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<String>,
) -> Result<Vec<Remote>, 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::<Vec<_>>(),
[&second.id, &first.id]
);
assert_eq!(
ordered
.iter()
.map(|remote| remote.priority)
.collect::<Vec<_>>(),
[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()
);
}
}
src/lib/bindings.ts
+4 -1
View File
@@ -32,6 +32,9 @@ async updateRemote(id: string, remote: RemoteInput) : Promise<Remote | null> {
async deleteRemote(id: string) : Promise<boolean> {
return await TAURI_INVOKE("delete_remote", { id });
},
async reorderRemotes(ids: string[]) : Promise<Remote[]> {
return await TAURI_INVOKE("reorder_remotes", { ids });
},
async getProfile() : Promise<User> {
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 }
src/lib/components/control-panel/network-panel.svelte
+53 -14
View File
@@ -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"
>
<div
class="grid grid-cols-[1fr_5.5rem] border-b border-base-300 bg-base-200 px-2 py-1 text-[10px] font-bold uppercase tracking-wide text-base-content/60"
class="grid grid-cols-[1fr_5.5rem_3.5rem] border-b border-base-300 bg-base-200 px-2 py-1 text-[10px] font-bold uppercase tracking-wide text-base-content/60"
>
<span>Server</span><span>Connection</span>
<span>Server</span><span>Connection</span><span>Priority</span>
</div>
<div
class="min-h-0 flex-1 overflow-y-auto"
role="listbox"
role="group"
aria-label="Configured servers"
>
{#if $remotes.length === 0}
@@ -138,17 +156,20 @@
</p>
</div>
{:else}
{#each $remotes as remote (remote.id)}
{#each $remotes as remote, index (remote.id)}
{@const state = statusFor(remote.id)}
<button
type="button"
role="option"
aria-selected={selectedId === remote.id}
class="grid w-full grid-cols-[1fr_5.5rem] items-center border-b border-base-200 px-2 py-1.5 text-left text-xs hover:bg-base-200 aria-selected:bg-primary aria-selected:text-primary-content"
onclick={() => (selectedId = remote.id)}
ondblclick={() => openEditWindow(remote.id)}
<div
class="grid w-full grid-cols-[1fr_5.5rem_3.5rem] items-center border-b border-base-200 px-2 py-1 text-xs hover:bg-base-200"
class:bg-primary={selectedId === remote.id}
class:text-primary-content={selectedId === remote.id}
>
<span class="min-w-0">
<button
type="button"
class="min-w-0 py-0.5 text-left"
aria-pressed={selectedId === remote.id}
onclick={() => (selectedId = remote.id)}
ondblclick={() => openEditWindow(remote.id)}
>
<strong class="block truncate"
>{remote.name ?? remote.address}</strong
>
@@ -157,7 +178,7 @@
? `:${remote.port}`
: ""}</span
>
</span>
</button>
<span class="flex items-center gap-1.5 capitalize">
<span
class:status-success={state === "connected"}
@@ -166,7 +187,25 @@
></span>
{state}
</span>
</button>
<span class="flex justify-end gap-0.5">
<button
class="btn btn-xs px-1"
type="button"
aria-label={`Increase priority for ${remote.name ?? remote.address}`}
title="Move up"
disabled={busy || index === 0}
onclick={() => moveRemote(remote.id, -1)}>↑</button
>
<button
class="btn btn-xs px-1"
type="button"
aria-label={`Decrease priority for ${remote.name ?? remote.address}`}
title="Move down"
disabled={busy || index === $remotes.length - 1}
onclick={() => moveRemote(remote.id, 1)}>↓</button
>
</span>
</div>
{/each}
{/if}
</div>