feat(lib): send and receive private messages

Wire private messaging through the client:

- MessageFactory::build_message_user (code 22) and build_message_acked
  (code 23);
- the incoming MessageUser handler now acknowledges freshly delivered
  messages and forwards them to the client instead of only logging them;
- ClientContext buffers received UserMessages;
- Client::send_private_message and Client::take_private_messages expose the
  feature, and UserMessage (with accessors) is re-exported from the crate.

Verified end to end against soulfind: a_private_message_is_delivered_between_users
logs in two clients and asserts one receives the other's message.

soulseek-rs-lib/src/actor/server_actor.rs

@@ -102,6 +102,37 @@ impl UserMessage {
             self.message
         );
     }
+
+    /// The server-assigned id of this message (used to acknowledge it).
+    #[must_use]
+    pub const fn id(&self) -> u32 {
+        self.id
+    }
+
+    /// Unix timestamp the server recorded for this message.
+    #[must_use]
+    pub const fn timestamp(&self) -> u32 {
+        self.timestamp
+    }
+
+    /// The username of the sender.
+    #[must_use]
+    pub fn username(&self) -> &str {
+        &self.username
+    }
+
+    /// The message body.
+    #[must_use]
+    pub fn message(&self) -> &str {
+        &self.message
+    }
+
+    /// Whether the server flagged this as freshly delivered (as opposed to a
+    /// message replayed because it was queued while the recipient was offline).
+    #[must_use]
+    pub const fn is_new(&self) -> bool {
+        self.new_message
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -129,6 +160,7 @@ pub enum ServerMessage {
         obfuscation_type: u32,
         obfuscated_port: u16,
     },
+    PrivateMessageReceived(UserMessage),
 }
 
 pub struct ServerActor {
@@ -418,6 +450,21 @@ impl ServerActor {
                     );
                 }
             }
+            ServerMessage::PrivateMessageReceived(user_message) => {
+                debug!(
+                    "[server] Private message from {}",
+                    user_message.username()
+                );
+                if let Err(e) = self
+                    .client_channel
+                    .send(ClientOperation::PrivateMessageReceived(user_message))
+                {
+                    error!(
+                        "[server] Error forwarding private message to client: {}",
+                        e
+                    );
+                }
+            }
             ServerMessage::ProcessRead => {
                 self.process_read();
             }

soulseek-rs-lib/src/client.rs

@@ -1,5 +1,7 @@
 use crate::actor::ActorHandle;
-use crate::actor::server_actor::{PeerAddress, ServerActor, ServerMessage};
+use crate::actor::server_actor::{
+    PeerAddress, ServerActor, ServerMessage, UserMessage,
+};
 use crate::download_store::{DownloadStore, collect_failed_tokens};
 use crate::types::{DownloadMetadata, DownloadStatus};
 use crate::utils::logger;
@@ -94,6 +96,7 @@ pub enum ClientOperation {
         place: u32,
     },
     SetServerSender(Sender<ServerMessage>),
+    PrivateMessageReceived(UserMessage),
 }
 pub struct ClientContext {
     pub peer_registry: Option<PeerRegistry>,
@@ -101,6 +104,7 @@ pub struct ClientContext {
     sender: Option<Sender<ClientOperation>>,
     server_sender: Option<Sender<ServerMessage>>,
     searches: HashMap<String, Search>,
+    private_messages: Vec<UserMessage>,
     actor_system: Arc<ActorSystem>,
 }
 impl Default for ClientContext {
@@ -307,10 +311,21 @@ impl ClientContext {
             sender: None,
             server_sender: None,
             searches: HashMap::new(),
+            private_messages: Vec::new(),
             downloads: DownloadStore::new(),
             actor_system,
         }
     }
+
+    /// Record a private message received from another user.
+    pub fn push_private_message(&mut self, message: UserMessage) {
+        self.private_messages.push(message);
+    }
+
+    /// Remove and return all buffered private messages.
+    pub fn take_private_messages(&mut self) -> Vec<UserMessage> {
+        std::mem::take(&mut self.private_messages)
+    }
 }
 pub struct Client {
     enable_listen: bool,
@@ -424,6 +439,40 @@ impl Client {
         }
     }
 
+    /// Send a private message to another user via the server.
+    ///
+    /// # Errors
+    /// Returns [`SoulseekRs::NotConnected`] if the client is not connected.
+    pub fn send_private_message(
+        &self,
+        username: &str,
+        message: &str,
+    ) -> Result<()> {
+        let handle = self
+            .server_handle
+            .as_ref()
+            .ok_or(SoulseekRs::NotConnected)?;
+        let msg = crate::message::server::MessageFactory::build_message_user(
+            username, message,
+        );
+        handle
+            .send(ServerMessage::SendMessage(msg))
+            .map_err(|_| SoulseekRs::NotConnected)?;
+        Ok(())
+    }
+
+    /// Remove and return all private messages received since the last call.
+    #[must_use]
+    pub fn take_private_messages(&self) -> Vec<UserMessage> {
+        match self.context.write_safe() {
+            Ok(mut ctx) => ctx.take_private_messages(),
+            Err(e) => {
+                error!("[client] take_private_messages: {}", e);
+                Vec::new()
+            }
+        }
+    }
+
     #[allow(dead_code)]
     pub fn remove_peer(&self, username: &str) {
         let context = match self.context.read_safe() {
@@ -1127,6 +1176,17 @@ impl Client {
                                     ),
                                 }
                             }
+                            ClientOperation::PrivateMessageReceived(
+                                user_message,
+                            ) => match client_context.write_safe() {
+                                Ok(mut ctx) => {
+                                    ctx.push_private_message(user_message);
+                                }
+                                Err(e) => error!(
+                                    "[client] PrivateMessageReceived write: {}",
+                                    e
+                                ),
+                            },
                         }
                     }
                     Err(e) => {

soulseek-rs-lib/src/lib.rs

@@ -20,7 +20,7 @@ pub mod prelude {
 }
 
 // Re-export commonly used types
-pub use actor::server_actor::PeerAddress;
+pub use actor::server_actor::{PeerAddress, UserMessage};
 pub use client::{Client, ClientSettings};
 pub use error::{Result, SoulseekRs};
 pub use types::{DownloadStatus, File, Search, SearchResult, Transfer};

soulseek-rs-lib/src/message/server/message_factory.rs

@@ -63,6 +63,23 @@ impl MessageFactory {
             .write_string(query)
             .clone()
     }
+    /// Build a private message (server code 22) to send to another user.
+    #[must_use]
+    pub fn build_message_user(username: &str, message: &str) -> Message {
+        Message::new()
+            .write_int32(22)
+            .write_string(username)
+            .write_string(message)
+            .clone()
+    }
+
+    /// Acknowledge a received private message (server code 23) so the server
+    /// stops re-delivering it.
+    #[must_use]
+    pub fn build_message_acked(id: u32) -> Message {
+        Message::new().write_int32(23).write_int32(id).clone()
+    }
+
     #[must_use]
     pub fn build_set_status_message(status_code: u32) -> Message {
         Message::new()
@@ -167,6 +184,25 @@ fn test_build_login_message() {
     assert_eq!(expect, message.get_data());
 }
 
+#[test]
+fn test_build_message_user() {
+    let message = MessageFactory::build_message_user("bob", "hi");
+    let expect: Vec<u8> = [
+        22, 0, 0, 0, // code
+        3, 0, 0, 0, 98, 111, 98, // username "bob"
+        2, 0, 0, 0, 104, 105, // message "hi"
+    ]
+    .to_vec();
+    assert_eq!(expect, message.get_data());
+}
+
+#[test]
+fn test_build_message_acked() {
+    let message = MessageFactory::build_message_acked(7);
+    let expect: Vec<u8> = [23, 0, 0, 0, 7, 0, 0, 0].to_vec();
+    assert_eq!(expect, message.get_data());
+}
+
 #[test]
 fn test_build_file_search_message() {
     let message = MessageFactory::build_file_search_message(12, "trance wax");

soulseek-rs-lib/src/message/server/message_user.rs

@@ -1,5 +1,6 @@
 use crate::actor::server_actor::{ServerMessage, UserMessage};
 use crate::info;
+use crate::message::server::MessageFactory;
 use crate::message::{Message, MessageHandler};
 
 use std::sync::mpsc::Sender;
@@ -11,7 +12,7 @@ impl MessageHandler<ServerMessage> for MessageUser {
         22
     }
 
-    fn handle(&self, message: &mut Message, _sender: Sender<ServerMessage>) {
+    fn handle(&self, message: &mut Message, sender: Sender<ServerMessage>) {
         let id = message.read_int32();
         let timestamp = message.read_int32();
         let username = message.read_string();
@@ -27,5 +28,17 @@ impl MessageHandler<ServerMessage> for MessageUser {
 
         info!("[MessageUser] User message received:{:?}", user_message);
         user_message.print();
+
+        // Acknowledge freshly delivered messages so the server does not keep
+        // re-sending them on every reconnect.
+        if new_message {
+            let _ = sender.send(ServerMessage::SendMessage(
+                MessageFactory::build_message_acked(id),
+            ));
+        }
+
+        // Surface the message to the client so it can be read via the API.
+        let _ =
+            sender.send(ServerMessage::PrivateMessageReceived(user_message));
     }
 }

soulseek-rs-lib/tests/e2e.rs

@@ -251,6 +251,42 @@ fn a_search_is_forwarded_to_a_connected_peer() {
     assert!(receiver.get_all_searches().contains_key(probe));
 }
 
+#[test]
+fn a_private_message_is_delivered_between_users() {
+    let server = server_or_skip!();
+
+    // Two logged-in users, one messages the other through the server.
+    let mut alice =
+        Client::with_settings(server.settings("e2e_alice_pm", "pw"));
+    let mut bob = Client::with_settings(server.settings("e2e_bob_pm", "pw"));
+    alice.connect().expect("alice connect");
+    bob.connect().expect("bob connect");
+    assert!(alice.login().expect("alice login"));
+    assert!(bob.login().expect("bob login"));
+
+    let body = "hello bob, this is alice";
+    alice
+        .send_private_message("e2e_bob_pm", body)
+        .expect("send private message");
+
+    // Delivery is asynchronous; poll Bob's inbox until the message arrives.
+    let deadline = Instant::now() + Duration::from_secs(5);
+    let mut received = Vec::new();
+    while Instant::now() < deadline {
+        received.extend(bob.take_private_messages());
+        if !received.is_empty() {
+            break;
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+
+    let message = received
+        .iter()
+        .find(|m| m.message() == body)
+        .expect("bob should receive alice's message");
+    assert_eq!(message.username(), "e2e_alice_pm");
+}
+
 #[test]
 fn login_succeeds_with_listener_enabled() {
     let server = server_or_skip!();