friends connection status system

src-server/src/network/interactions.rs

@@ -10,7 +10,7 @@ use wyd_common::{
     MAX_INTERACTION_PAYLOAD_BYTES, ServerMessage, interaction_bytes,
 };
 
-use super::{Clients, verify};
+use super::{Clients, are_mutual_friends, verify};
 
 pub(super) async fn relay(
     clients: &Clients,
@@ -28,10 +28,7 @@ pub(super) async fn relay(
             .filter(|client| client.connection_id == connection_id)?;
         let recipient = clients
             .get(recipient_id)
-            .filter(|recipient| {
-                source.friends.iter().any(|friend| friend == recipient_id)
-                    && recipient.friends.iter().any(|friend| friend == public_key)
-            })
+            .filter(|recipient| are_mutual_friends(public_key, source, recipient_id, recipient))
             .map(|recipient| recipient.sender.clone());
         (source.key, recipient)
     };

src-server/src/network/mod.rs

@@ -14,11 +14,11 @@ use futures_util::{SinkExt, StreamExt};
 use tokio::sync::{Mutex, mpsc};
 use uuid::Uuid;
 use wyd_common::{
-    ClientMessage, Profile, ServerMessage, friends_bytes, message_bytes, profile_bytes,
-    register_bytes,
+    ClientMessage, Profile, ServerMessage, message_bytes, profile_bytes, register_bytes,
 };
 
 mod interactions;
+mod presence;
 
 type Clients = Arc<Mutex<HashMap<String, Client>>>;
 
@@ -31,6 +31,11 @@ struct Client {
     sender: mpsc::Sender<Message>,
 }
 
+fn are_mutual_friends(source_id: &str, source: &Client, friend_id: &str, friend: &Client) -> bool {
+    source.friends.iter().any(|id| id == friend_id)
+        && friend.friends.iter().any(|id| id == source_id)
+}
+
 pub fn routes() -> Router {
     Router::new()
         .route("/v1/ws", get(upgrade))
@@ -79,20 +84,25 @@ async fn connected(mut socket: WebSocket, clients: Clients) {
     let connection_id = Uuid::new_v4();
     let public_key = profile.id.clone();
     let (sender, mut outgoing) = mpsc::channel(32);
-    clients.lock().await.insert(
-        public_key.clone(),
-        Client {
-            connection_id,
-            key,
-            profile,
-            friends,
-            sender,
-        },
-    );
+    let previous_friends = clients
+        .lock()
+        .await
+        .insert(
+            public_key.clone(),
+            Client {
+                connection_id,
+                key,
+                profile,
+                friends,
+                sender,
+            },
+        )
+        .map(|client| client.friends);
     if send(&mut socket, &ServerMessage::Registered).await.is_err() {
-        remove(&clients, &public_key, connection_id).await;
+        presence::disconnected(&clients, &public_key, connection_id).await;
         return;
     }
+    presence::connected(&clients, &public_key, previous_friends).await;
 
     let (mut writer, mut reader) = socket.split();
     let mut ping = tokio::time::interval(Duration::from_secs(20));
@@ -143,7 +153,17 @@ async fn connected(mut socket: WebSocket, clients: Clients) {
                         }
                     }
                     Ok(ClientMessage::FriendsUpdated { friends, signature }) => {
-                        if !update_friends(&clients, &public_key, connection_id, friends, &signature).await {
+                        let Some(friend_ids) = presence::update_friends(
+                            &clients,
+                            &public_key,
+                            connection_id,
+                            friends,
+                            &signature,
+                        ).await else {
+                            break;
+                        };
+                        let message = ServerMessage::FriendStatuses { friend_ids };
+                        if writer.send(Message::Text(serde_json::to_string(&message).unwrap().into())).await.is_err() {
                             break;
                         }
                     }
@@ -156,6 +176,15 @@ async fn connected(mut socket: WebSocket, clients: Clients) {
                             break;
                         }
                     }
+                    Ok(ClientMessage::SyncFriendStatuses) => {
+                        let Some(friend_ids) = presence::snapshot(&clients, &public_key, connection_id).await else {
+                            break;
+                        };
+                        let message = ServerMessage::FriendStatuses { friend_ids };
+                        if writer.send(Message::Text(serde_json::to_string(&message).unwrap().into())).await.is_err() {
+                            break;
+                        }
+                    }
                     _ => break,
                 }
                 Some(Ok(Message::Ping(data))) => {
@@ -167,7 +196,7 @@ async fn connected(mut socket: WebSocket, clients: Clients) {
         }
     }
 
-    remove(&clients, &public_key, connection_id).await;
+    presence::disconnected(&clients, &public_key, connection_id).await;
 }
 
 async fn update_profile(
@@ -209,27 +238,6 @@ async fn update_profile(
     true
 }
 
-async fn update_friends(
-    clients: &Clients,
-    public_key: &str,
-    connection_id: Uuid,
-    friends: Vec<String>,
-    signature: &str,
-) -> bool {
-    let mut clients = clients.lock().await;
-    let Some(client) = clients
-        .get_mut(public_key)
-        .filter(|client| client.connection_id == connection_id)
-    else {
-        return false;
-    };
-    if !verify(&client.key, &friends_bytes(&friends), signature) {
-        return false;
-    }
-    client.friends = friends;
-    true
-}
-
 async fn friend_profiles(
     clients: &Clients,
     public_key: &str,
@@ -270,8 +278,7 @@ async fn relay_live_data(
         .iter()
         .filter(|(recipient_id, recipient)| {
             recipient_id.as_str() != public_key
-                && source.friends.iter().any(|friend| friend == *recipient_id)
-                && recipient.friends.iter().any(|friend| friend == public_key)
+                && are_mutual_friends(public_key, source, recipient_id, recipient)
         })
         .map(|(_, recipient)| recipient.sender.clone())
         .collect();
@@ -291,16 +298,6 @@ async fn relay_live_data(
     true
 }
 
-async fn remove(clients: &Clients, public_key: &str, connection_id: Uuid) {
-    let mut clients = clients.lock().await;
-    if clients
-        .get(public_key)
-        .is_some_and(|client| client.connection_id == connection_id)
-    {
-        clients.remove(public_key);
-    }
-}
-
 async fn send(socket: &mut WebSocket, message: &ServerMessage) -> Result<(), axum::Error> {
     socket
         .send(Message::Text(
@@ -328,6 +325,7 @@ fn verify(key: &VerifyingKey, bytes: &[u8], signature: &str) -> bool {
 #[cfg(test)]
 mod tests {
     use ed25519_dalek::{Signer, SigningKey};
+    use wyd_common::friends_bytes;
 
     use super::*;
 
@@ -492,7 +490,7 @@ mod tests {
             URL_SAFE_NO_PAD.encode(signing_key.sign(&friends_bytes(&friends)).to_bytes());
 
         assert!(
-            update_friends(
+            presence::update_friends(
                 &clients,
                 &public_key,
                 connection_id,
@@ -500,6 +498,7 @@ mod tests {
                 &signature,
             )
             .await
+            .is_some()
         );
         assert_eq!(
             clients.lock().await.get(&public_key).unwrap().friends,
@@ -565,6 +564,108 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn presence_sync_and_disconnect_broadcast_require_mutual_friendship() {
+        let signing_key = SigningKey::from_bytes(&[7; 32]);
+        let connection_id = Uuid::new_v4();
+        let (source_sender, _source_receiver) = mpsc::channel(4);
+        let (mutual_sender, mut mutual_receiver) = mpsc::channel(4);
+        let (one_way_sender, mut one_way_receiver) = mpsc::channel(4);
+        let clients = Clients::default();
+        let mut registry = clients.lock().await;
+        registry.insert(
+            "source".to_owned(),
+            Client {
+                connection_id,
+                key: signing_key.verifying_key(),
+                profile: Profile {
+                    id: "source".to_owned(),
+                    display_name: "Source".to_owned(),
+                },
+                friends: vec!["mutual".to_owned()],
+                sender: source_sender,
+            },
+        );
+        registry.insert(
+            "mutual".to_owned(),
+            Client {
+                connection_id: Uuid::new_v4(),
+                key: signing_key.verifying_key(),
+                profile: Profile {
+                    id: "mutual".to_owned(),
+                    display_name: "Mutual".to_owned(),
+                },
+                friends: vec!["source".to_owned()],
+                sender: mutual_sender,
+            },
+        );
+        registry.insert(
+            "one-way".to_owned(),
+            Client {
+                connection_id: Uuid::new_v4(),
+                key: signing_key.verifying_key(),
+                profile: Profile {
+                    id: "one-way".to_owned(),
+                    display_name: "One way".to_owned(),
+                },
+                friends: vec!["source".to_owned()],
+                sender: one_way_sender,
+            },
+        );
+        drop(registry);
+
+        assert_eq!(
+            presence::snapshot(&clients, "source", connection_id)
+                .await
+                .expect("current session"),
+            ["mutual"]
+        );
+        assert!(
+            presence::snapshot(&clients, "source", Uuid::new_v4())
+                .await
+                .is_none()
+        );
+
+        presence::broadcast_status(&clients, "source", true).await;
+        assert_friend_status(mutual_receiver.recv().await, "source", true);
+        assert!(one_way_receiver.try_recv().is_err());
+
+        let no_friends = Vec::new();
+        let signature =
+            URL_SAFE_NO_PAD.encode(signing_key.sign(&friends_bytes(&no_friends)).to_bytes());
+        assert_eq!(
+            presence::update_friends(&clients, "source", connection_id, no_friends, &signature,)
+                .await,
+            Some(Vec::new())
+        );
+        assert_friend_status(mutual_receiver.recv().await, "source", false);
+
+        let mutual_friends = vec!["mutual".to_owned()];
+        let signature =
+            URL_SAFE_NO_PAD.encode(signing_key.sign(&friends_bytes(&mutual_friends)).to_bytes());
+        assert_eq!(
+            presence::update_friends(
+                &clients,
+                "source",
+                connection_id,
+                mutual_friends,
+                &signature,
+            )
+            .await,
+            Some(vec!["mutual".to_owned()])
+        );
+        assert_friend_status(mutual_receiver.recv().await, "source", true);
+
+        presence::disconnected(&clients, "source", Uuid::new_v4()).await;
+        assert!(clients.lock().await.contains_key("source"));
+        assert!(mutual_receiver.try_recv().is_err());
+
+        presence::disconnected(&clients, "source", connection_id).await;
+        assert_friend_status(mutual_receiver.recv().await, "source", false);
+        assert!(one_way_receiver.try_recv().is_err());
+        assert!(!clients.lock().await.contains_key("source"));
+    }
+
     #[tokio::test]
     async fn live_data_relay_requires_valid_session_signature_and_mutual_friendship() {
         let signing_key = SigningKey::from_bytes(&[7; 32]);
@@ -669,4 +770,17 @@ mod tests {
         );
         assert!(mutual_receiver.try_recv().is_err());
     }
+
+    fn assert_friend_status(message: Option<Message>, expected_id: &str, expected_online: bool) {
+        let Message::Text(message) = message.expect("friend receives status") else {
+            panic!("expected text status");
+        };
+        let ServerMessage::FriendStatusChanged { friend_id, online } =
+            serde_json::from_str(&message).expect("decode friend status")
+        else {
+            panic!("expected friend status");
+        };
+        assert_eq!(friend_id, expected_id);
+        assert_eq!(online, expected_online);
+    }
 }

src-server/src/network/presence.rs

@@ -0,0 +1,179 @@
+use axum::extract::ws::Message;
+use tokio::sync::mpsc;
+use uuid::Uuid;
+use wyd_common::{ServerMessage, friends_bytes};
+
+use super::{Clients, are_mutual_friends, verify};
+
+pub(super) async fn connected(
+    clients: &Clients,
+    public_key: &str,
+    previous_friends: Option<Vec<String>>,
+) {
+    if let Some(previous_friends) = previous_friends {
+        broadcast_friendship_changes(clients, public_key, &previous_friends).await;
+    } else {
+        broadcast_status(clients, public_key, true).await;
+    }
+}
+
+pub(super) async fn update_friends(
+    clients: &Clients,
+    public_key: &str,
+    connection_id: Uuid,
+    friends: Vec<String>,
+    signature: &str,
+) -> Option<Vec<String>> {
+    let mut clients = clients.lock().await;
+    let previous_friends = {
+        let client = clients
+            .get_mut(public_key)
+            .filter(|client| client.connection_id == connection_id)?;
+        if !verify(&client.key, &friends_bytes(&friends), signature) {
+            return None;
+        }
+        std::mem::replace(&mut client.friends, friends.clone())
+    };
+
+    let mut notifications = Vec::new();
+    let mut online_friend_ids = Vec::new();
+    for (friend_id, friend) in clients.iter().filter(|(id, _)| id.as_str() != public_key) {
+        let was_visible = mutual_friends(public_key, &previous_friends, friend_id, &friend.friends);
+        let is_visible = mutual_friends(public_key, &friends, friend_id, &friend.friends);
+        if is_visible {
+            online_friend_ids.push(friend_id.clone());
+        }
+        if was_visible != is_visible {
+            notifications.push((friend.sender.clone(), public_key.to_owned(), is_visible));
+        }
+    }
+    drop(clients);
+
+    send_notifications(notifications).await;
+    online_friend_ids.sort_unstable();
+    Some(online_friend_ids)
+}
+
+pub(super) async fn snapshot(
+    clients: &Clients,
+    public_key: &str,
+    connection_id: Uuid,
+) -> Option<Vec<String>> {
+    let clients = clients.lock().await;
+    let source = clients
+        .get(public_key)
+        .filter(|client| client.connection_id == connection_id)?;
+    let mut friend_ids = clients
+        .iter()
+        .filter(|(friend_id, friend)| {
+            friend_id.as_str() != public_key && visible_to(public_key, source, friend_id, friend)
+        })
+        .map(|(friend_id, _)| friend_id.clone())
+        .collect::<Vec<_>>();
+    friend_ids.sort_unstable();
+    Some(friend_ids)
+}
+
+pub(super) async fn broadcast_status(clients: &Clients, public_key: &str, online: bool) {
+    let recipients = {
+        let clients = clients.lock().await;
+        let Some(source) = clients.get(public_key) else {
+            return;
+        };
+        clients
+            .iter()
+            .filter(|(friend_id, friend)| {
+                friend_id.as_str() != public_key
+                    && visible_to(public_key, source, friend_id, friend)
+            })
+            .map(|(_, friend)| friend.sender.clone())
+            .collect::<Vec<_>>()
+    };
+    let message = status_message(public_key.to_owned(), online);
+    for recipient in recipients {
+        let _ = recipient.send(message.clone()).await;
+    }
+}
+
+pub(super) async fn disconnected(clients: &Clients, public_key: &str, connection_id: Uuid) {
+    let recipients = {
+        let mut clients = clients.lock().await;
+        if !clients
+            .get(public_key)
+            .is_some_and(|client| client.connection_id == connection_id)
+        {
+            return;
+        }
+        let source = clients
+            .remove(public_key)
+            .expect("checked registered client");
+        clients
+            .iter()
+            .filter(|(friend_id, friend)| visible_to(public_key, &source, friend_id, friend))
+            .map(|(_, friend)| friend.sender.clone())
+            .collect::<Vec<_>>()
+    };
+    let message = status_message(public_key.to_owned(), false);
+    for recipient in recipients {
+        let _ = recipient.send(message.clone()).await;
+    }
+}
+
+async fn broadcast_friendship_changes(
+    clients: &Clients,
+    public_key: &str,
+    previous_friends: &[String],
+) {
+    let notifications = {
+        let clients = clients.lock().await;
+        let Some(source) = clients.get(public_key) else {
+            return;
+        };
+        clients
+            .iter()
+            .filter(|(friend_id, _)| friend_id.as_str() != public_key)
+            .filter_map(|(friend_id, friend)| {
+                let was_visible =
+                    mutual_friends(public_key, previous_friends, friend_id, &friend.friends);
+                let is_visible = visible_to(public_key, source, friend_id, friend);
+                (was_visible != is_visible)
+                    .then(|| (friend.sender.clone(), public_key.to_owned(), is_visible))
+            })
+            .collect::<Vec<_>>()
+    };
+    send_notifications(notifications).await;
+}
+
+async fn send_notifications(notifications: Vec<(mpsc::Sender<Message>, String, bool)>) {
+    for (recipient, friend_id, online) in notifications {
+        let _ = recipient.send(status_message(friend_id, online)).await;
+    }
+}
+
+fn visible_to(
+    source_id: &str,
+    source: &super::Client,
+    viewer_id: &str,
+    viewer: &super::Client,
+) -> bool {
+    // Apply the source profile's per-friend online-visibility preference here.
+    are_mutual_friends(source_id, source, viewer_id, viewer)
+}
+
+fn mutual_friends(
+    source_id: &str,
+    source_friends: &[String],
+    viewer_id: &str,
+    viewer_friends: &[String],
+) -> bool {
+    source_friends.iter().any(|friend| friend == viewer_id)
+        && viewer_friends.iter().any(|friend| friend == source_id)
+}
+
+fn status_message(friend_id: String, online: bool) -> Message {
+    Message::Text(
+        serde_json::to_string(&ServerMessage::FriendStatusChanged { friend_id, online })
+            .expect("friend status serializes")
+            .into(),
+    )
+}