Inital Commit (Errors, I need to move through devices)

Signed-off-by: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com>

src/rho/client_connection.rs

@@ -0,0 +1,515 @@
+use futures::SinkExt;
+use http::header::AUTHORIZATION;
+use std::{
+    collections::HashMap,
+    sync::{Arc, Weak},
+};
+use tokio::sync::{Mutex, RwLock};
+use tokio_tungstenite::{WebSocketStream, tungstenite::Message};
+use tungstenite::Utf8Bytes;
+use uuid::Uuid;
+
+use super::{rho_connection::RhoConnection, rho_manager};
+use crate::{
+    auth::auth_connector,
+    // calls::call_manager::CallManager,
+    data::{
+        communication::{CommunicationType, CommunicationValue, DataTypes},
+        user::{User, UserStatus},
+    },
+    omega::omega_connection::OmegaConnection,
+};
+
+/// ClientConnection represents a WebSocket connection from a client device
+pub struct ClientConnection {
+    /// WebSocket session
+    session: Arc<Mutex<WebSocketStream<tokio::net::TcpStream>>>,
+    /// User ID associated with this client
+    user_id: Arc<RwLock<Option<Uuid>>>,
+    /// Whether this connection has been identified/authenticated
+    identified: Arc<RwLock<bool>>,
+    /// Ping latency tracking
+    ping: Arc<RwLock<i64>>,
+    /// Weak reference to RhoConnection to avoid circular references
+    rho_connection: Arc<RwLock<Option<Weak<RhoConnection>>>>,
+    /// List of user IDs this client is interested in receiving updates about
+    interested_users: Arc<RwLock<Vec<Uuid>>>,
+}
+
+impl ClientConnection {
+    /// Create a new ClientConnection
+    pub fn new(session: WebSocketStream<tokio::net::TcpStream>) -> Arc<Self> {
+        Arc::new(Self {
+            session: Arc::new(Mutex::new(session)),
+            user_id: Arc::new(RwLock::new(None)),
+            identified: Arc::new(RwLock::new(false)),
+            ping: Arc::new(RwLock::new(-1)),
+            rho_connection: Arc::new(RwLock::new(None)),
+            interested_users: Arc::new(RwLock::new(Vec::new())),
+        })
+    }
+
+    /// Get the user ID
+    pub async fn get_user_id(&self) -> Option<Uuid> {
+        *self.user_id.read().await
+    }
+
+    /// Check if connection is identified
+    pub async fn is_identified(&self) -> bool {
+        *self.identified.read().await
+    }
+
+    /// Get current ping
+    pub async fn get_ping(&self) -> i64 {
+        *self.ping.read().await
+    }
+
+    /// Set the RhoConnection reference
+    pub async fn set_rho_connection(&self, rho_connection: Weak<RhoConnection>) {
+        let mut rho_ref = self.rho_connection.write().await;
+        *rho_ref = Some(rho_connection);
+    }
+
+    /// Get RhoConnection if available
+    pub async fn get_rho_connection(&self) -> Option<Arc<RhoConnection>> {
+        let rho_ref = self.rho_connection.read().await;
+        if let Some(weak_ref) = rho_ref.as_ref() {
+            weak_ref.upgrade()
+        } else {
+            None
+        }
+    }
+
+    /// Send a string message to the client
+    pub async fn send_message_str(&self, message: &str) {
+        let mut session = self.session.lock().await;
+        if let Err(e) = session
+            .send(Message::Text(Utf8Bytes::from(message.to_string())))
+            .await
+        {
+            eprintln!("Failed to send message to client: {}", e);
+        }
+    }
+
+    /// Send a CommunicationValue to the client
+    pub async fn send_message(&self, cv: &CommunicationValue) {
+        self.send_message_str(&cv.to_json().to_string()).await;
+    }
+
+    /// Handle incoming message from client
+    pub async fn handle_message(self: Arc<Self>, message: String) {
+        let cv = CommunicationValue::from_json(&message);
+
+        // Handle identification
+        if cv.is_type(CommunicationType::identification) && !self.is_identified().await {
+            self.handle_identification(Arc::clone(&self), cv).await;
+            return;
+        }
+
+        if !self.is_identified().await {
+            return;
+        }
+
+        // Handle ping
+        if cv.is_type(CommunicationType::ping) {
+            self.handle_ping(cv).await;
+            return;
+        }
+
+        // Handle client status changes
+        if cv.is_type(CommunicationType::client_changed) {
+            self.handle_client_changed(cv).await;
+            return;
+        }
+
+        // Handle call invites
+        if cv.is_type(CommunicationType::call_invite) {
+            self.handle_call_invite(cv).await;
+            return;
+        }
+
+        // Handle get call requests
+        if cv.is_type(CommunicationType::get_call) {
+            self.handle_get_call(cv).await;
+            return;
+        }
+
+        // Forward other messages to Iota
+        self.forward_to_iota(cv).await;
+    }
+
+    /// Handle identification message
+    async fn handle_identification(&self, sarc: Arc<ClientConnection>, mut cv: CommunicationValue) {
+        // Extract user ID
+        let user_id = match cv.get_data(DataTypes::user_id) {
+            Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(&cv.get_id(), CommunicationType::error)
+                        .await;
+                    return;
+                }
+            },
+            None => {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        // Find RhoConnection for this user
+        let rho_connection = match rho_manager::get_rho_con_for_user(user_id).await {
+            Some(rho) => rho,
+            None => {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        // Validate private key
+        if let Some(private_key_hash) = cv.get_data(DataTypes::private_key_hash) {
+            let is_valid =
+                auth_connector::is_private_key_valid(user_id, &private_key_hash.to_string()).await;
+
+            if !is_valid {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        } else {
+            self.send_error_response(&cv.get_id(), CommunicationType::error)
+                .await;
+            return;
+        }
+
+        // Set identification data
+        {
+            let mut user_id_guard = self.user_id.write().await;
+            *user_id_guard = Some(user_id);
+        }
+        {
+            let mut identified_guard = self.identified.write().await;
+            *identified_guard = true;
+        }
+
+        // Set up RhoConnection reference
+        self.set_rho_connection(Arc::downgrade(&rho_connection))
+            .await;
+
+        // Add this client to the RhoConnection
+        rho_connection.add_client_connection(Arc::from(sarc)).await;
+
+        // Send success response
+        let response = CommunicationValue::new(CommunicationType::identification_response)
+            .with_id(cv.get_id());
+        self.send_message(&response).await;
+    }
+
+    /// Handle ping message
+    async fn handle_ping(&self, mut cv: CommunicationValue) {
+        // Update our ping if provided
+        if let Some(last_ping) = cv.get_data(DataTypes::last_ping) {
+            if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
+                let mut ping_guard = self.ping.write().await;
+                *ping_guard = ping_val;
+            }
+        }
+
+        // Get Iota ping from RhoConnection
+        let iota_ping = if let Some(rho_conn) = self.get_rho_connection().await {
+            rho_conn.get_iota_connection().get_ping().await
+        } else {
+            -1
+        };
+
+        // Send pong response
+        let response = CommunicationValue::new(CommunicationType::pong)
+            .with_id(cv.get_id())
+            .add_data_str(DataTypes::ping_iota, iota_ping.to_string());
+
+        self.send_message(&response).await;
+    }
+
+    /// Handle client status change
+    async fn handle_client_changed(&self, mut cv: CommunicationValue) {
+        if let Some(user_id) = self.get_user_id().await {
+            if let Some(status_str) = cv.get_data(DataTypes::user_state) {
+                // Parse user status - this would need to be implemented properly
+                let user_status = UserStatus::online; // placeholder
+                if let Some(rho_conn) = self.get_rho_connection().await {
+                    OmegaConnection::client_changed(
+                        rho_conn.get_iota_id().await,
+                        user_id,
+                        user_status,
+                    );
+                }
+            }
+        }
+    }
+
+    /// Handle call invite
+    async fn handle_call_invite(&self, mut cv: CommunicationValue) {
+        let receiver_id = match cv.get_data(DataTypes::receiver_id) {
+            Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(&cv.get_id(), CommunicationType::error)
+                        .await;
+                    return;
+                }
+            },
+            None => {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        let call_id = match cv.get_data(DataTypes::call_id) {
+            Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(&cv.get_id(), CommunicationType::error)
+                        .await;
+                    return;
+                }
+            },
+            None => {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        let call_secret_sha = cv
+            .get_data(DataTypes::call_secret_sha)
+            .map(|s| s.to_string());
+        let call_secret = cv.get_data(DataTypes::call_secret).map(|s| s.to_string());
+
+        if call_secret_sha.is_none() || call_secret.is_none() {
+            self.send_error_response(&cv.get_id(), CommunicationType::error)
+                .await;
+            return;
+        }
+
+        // Get call group - placeholder implementation
+        // let call_group = CallManager::get_call_group(call_id, &call_secret_sha.unwrap(), true).await;
+        // if call_group.is_none() {
+        //     self.send_error_response(&cv.get_id(), CommunicationType::error)
+        //         .await;
+        //     return;
+        // }
+
+        // Find target RhoConnection
+        let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
+            Some(rho) => rho,
+            None => {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        // Get sender user ID
+        let sender_id = match self.get_user_id().await {
+            Some(id) => id,
+            None => return,
+        };
+
+        // Create and send call distribution message
+        let distribute = CommunicationValue::new(CommunicationType::new_call)
+            .with_receiver(receiver_id)
+            .with_sender(sender_id)
+            .add_data_str(DataTypes::call_id, call_id.to_string())
+            .add_data_str(DataTypes::receiver_id, receiver_id.to_string())
+            .add_data_str(DataTypes::sender_id, sender_id.to_string())
+            .add_data_str(DataTypes::call_secret, call_secret.unwrap());
+
+        target_rho.message_iota_to_client(distribute).await;
+
+        // Handle call group invitation logic here
+        // This would require implementing CallGroup::Caller and related functionality
+
+        // Send success response
+        let response = CommunicationValue::new(CommunicationType::call_invite).with_id(cv.get_id());
+        self.send_message(&response).await;
+    }
+
+    /// Handle get call request
+    async fn handle_get_call(&self, mut cv: CommunicationValue) {
+        let user_id = match self.get_user_id().await {
+            Some(id) => id,
+            None => return,
+        };
+
+        let call_id = match cv.get_data(DataTypes::call_id) {
+            Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(&cv.get_id(), CommunicationType::error)
+                        .await;
+                    return;
+                }
+            },
+            None => {
+                self.send_error_response(&cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        let call_secret_sha = cv
+            .get_data(DataTypes::call_secret_sha)
+            .map(|s| s.to_string());
+
+        if call_secret_sha.is_none() {
+            self.send_error_response(&cv.get_id(), CommunicationType::error)
+                .await;
+            return;
+        }
+
+        // Get call group
+        let mut response = CommunicationValue::new(CommunicationType::get_call)
+            .with_id(cv.get_id())
+            .with_receiver(user_id);
+
+        // Placeholder call group logic
+        // if let Some(call_group) = CallManager::get_call_group(call_id, &call_secret_sha.unwrap(), true).await {
+        //     response = response
+        //         .add_data_str(DataTypes::call_state, call_group.call_state.to_string())
+        //         .add_data_str(DataTypes::start_date, call_group.started_at.to_string());
+        //
+        //     if call_group.ended_at != 0 {
+        //         response = response.add_data_str(DataTypes::end_date, call_group.ended_at.to_string());
+        //     }
+        // } else {
+        response = response.add_data_str(DataTypes::call_state, "DESTROYED".to_string());
+        // }
+
+        self.send_message(&response).await;
+    }
+
+    /// Forward message to Iota
+    async fn forward_to_iota(&self, mut cv: CommunicationValue) {
+        if let Some(user_id) = self.get_user_id().await {
+            if let Some(rho_conn) = self.get_rho_connection().await {
+                let updated_cv = cv.with_sender(user_id);
+                rho_conn.message_to_iota(updated_cv).await;
+            }
+        }
+    }
+
+    /// Send error response
+    async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) {
+        let error = CommunicationValue::new(error_type).with_id(*message_id);
+        self.send_message(&error).await;
+    }
+
+    /// Close the connection
+    pub async fn close(&self) {
+        let mut session = self.session.lock().await;
+        let _ = session.close(None).await;
+    }
+
+    /// Set interested users list
+    pub async fn set_interested_users(&self, interested_ids: Vec<Uuid>) {
+        let mut interested_guard = self.interested_users.write().await;
+        *interested_guard = interested_ids;
+    }
+
+    /// Check if interested in a user and send notification
+    pub async fn are_you_interested(&self, user: &User) {
+        let interested_guard = self.interested_users.read().await;
+        if interested_guard.contains(&user.user_id) {
+            let notification = CommunicationValue::new(CommunicationType::client_changed)
+                .add_data_str(DataTypes::user_id, user.user_id.to_string())
+                .add_data_str(
+                    DataTypes::user_state,
+                    format!("{:?}", user.status.to_string()),
+                );
+
+            self.send_message(&notification).await;
+        }
+    }
+
+    /// Handle connection close
+    pub async fn handle_close(&self) {
+        if self.is_identified().await {
+            if let Some(user_id) = self.get_user_id().await {
+                if let Some(rho_conn) = rho_manager::get_rho_con_for_user(user_id).await {
+                    rho_conn
+                        .close_client_connection(Arc::new(self.clone()))
+                        .await;
+                }
+            }
+        }
+    }
+}
+
+// Implement Clone to make it easier to work with Arc<ClientConnection>
+impl Clone for ClientConnection {
+    fn clone(&self) -> Self {
+        Self {
+            session: Arc::clone(&self.session),
+            user_id: Arc::clone(&self.user_id),
+            identified: Arc::clone(&self.identified),
+            ping: Arc::clone(&self.ping),
+            rho_connection: Arc::clone(&self.rho_connection),
+            interested_users: Arc::clone(&self.interested_users),
+        }
+    }
+}
+
+impl std::fmt::Debug for ClientConnection {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ClientConnection")
+            .field("user_id", &"[async]")
+            .field("identified", &"[async]")
+            .field("ping", &"[async]")
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::sync::Mutex;
+
+    #[tokio::test]
+    async fn test_client_connection_creation() {
+        // Mock session - in real implementation this would be a proper WebSocket stream
+        let mock_session = tokio_tungstenite::WebSocketStream::from_raw_socket(
+            tokio::net::TcpStream::connect("127.0.0.1:0")
+                .await
+                .unwrap_or_else(|_| {
+                    // This is just for testing, create a dummy stream
+                    panic!("Cannot create test stream")
+                }),
+            tokio_tungstenite::tungstenite::protocol::Role::Client,
+            None,
+        )
+        .await;
+
+        // This test would fail in practice due to the mock stream
+        // but shows the intended API
+        // let client_conn = ClientConnection::new(mock_session);
+
+        // assert!(!client_conn.is_identified().await);
+        // assert_eq!(client_conn.get_ping().await, -1);
+        // assert!(client_conn.get_user_id().await.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_interested_users() {
+        // This would also need a proper mock setup
+        // but shows the intended functionality
+
+        // let client_conn = ClientConnection::new(mock_session);
+        // let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()];
+
+        // client_conn.set_interested_users(user_ids.clone()).await;
+        //
+        // // Test would verify that the interested users are stored correctly
+    }
+}

src/rho/iota_connection.rs

@@ -0,0 +1,345 @@
+use futures::SinkExt;
+use json::{JsonValue, number::Number};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Weak},
+};
+use tokio::sync::{Mutex, RwLock};
+use tokio_tungstenite::{WebSocketStream, tungstenite::Message};
+use tungstenite::Utf8Bytes;
+use uuid::Uuid;
+
+use super::{rho_connection::RhoConnection, rho_manager};
+use crate::{
+    auth::auth_connector,
+    // calls::call_manager::CallManager,
+    data::{
+        communication::{CommunicationType, CommunicationValue, DataTypes},
+        user::User,
+    },
+    omega::omega_connection::OmegaConnection,
+};
+
+/// IotaConnection represents a WebSocket connection from an Iota device
+pub struct IotaConnection {
+    session: Arc<Mutex<WebSocketStream<tokio::net::TcpStream>>>,
+    iota_id: Arc<RwLock<Uuid>>,
+    user_ids: Arc<RwLock<Vec<Uuid>>>,
+    identified: Arc<RwLock<bool>>,
+    ping: Arc<RwLock<i64>>,
+    rho_connection: Arc<RwLock<Option<Weak<RhoConnection>>>>,
+}
+
+impl IotaConnection {
+    /// Create a new IotaConnection
+    pub fn new(session: WebSocketStream<tokio::net::TcpStream>) -> Arc<Self> {
+        Arc::new(Self {
+            session: Arc::new(Mutex::new(session)),
+            iota_id: Arc::new(RwLock::new(Uuid::nil())),
+            user_ids: Arc::new(RwLock::new(Vec::new())),
+            identified: Arc::new(RwLock::new(false)),
+            ping: Arc::new(RwLock::new(0)),
+            rho_connection: Arc::new(RwLock::new(None)),
+        })
+    }
+
+    /// Create with known IDs (for testing)
+    pub fn new_with_ids(
+        iota_id: Uuid,
+        user_ids: Vec<Uuid>,
+        session: Arc<Mutex<WebSocketStream<tokio::net::TcpStream>>>,
+    ) -> Arc<Self> {
+        Arc::new(Self {
+            session,
+            iota_id: Arc::new(RwLock::new(iota_id)),
+            user_ids: Arc::new(RwLock::new(user_ids)),
+            identified: Arc::new(RwLock::new(true)),
+            ping: Arc::new(RwLock::new(0)),
+            rho_connection: Arc::new(RwLock::new(None)),
+        })
+    }
+
+    /// Get the Iota ID
+    pub async fn get_iota_id(&self) -> Uuid {
+        *self.iota_id.read().await
+    }
+
+    /// Get the user IDs
+    pub async fn get_user_ids(&self) -> Vec<Uuid> {
+        self.user_ids.read().await.clone()
+    }
+
+    /// Check if connection is identified
+    pub async fn is_identified(&self) -> bool {
+        *self.identified.read().await
+    }
+
+    /// Get current ping
+    pub async fn get_ping(&self) -> i64 {
+        *self.ping.read().await
+    }
+
+    /// Set the RhoConnection reference
+    pub async fn set_rho_connection(&self, rho_connection: Weak<RhoConnection>) {
+        let mut rho_ref = self.rho_connection.write().await;
+        *rho_ref = Some(rho_connection);
+    }
+
+    /// Get RhoConnection if available
+    pub async fn get_rho_connection(&self) -> Option<Arc<RhoConnection>> {
+        let rho_ref = self.rho_connection.read().await;
+        if let Some(weak_ref) = rho_ref.as_ref() {
+            weak_ref.upgrade()
+        } else {
+            None
+        }
+    }
+
+    /// Send a message to the Iota
+    pub async fn send_message_str(&self, message: &str) {
+        let mut session = self.session.lock().await;
+        let _ = session
+            .send(Message::Text(Utf8Bytes::from(message.to_string())))
+            .await;
+    }
+
+    /// Send a CommunicationValue to the Iota
+    pub async fn send_message(&self, cv: CommunicationValue) {
+        self.send_message_str(&cv.to_json().to_string()).await;
+    }
+
+    /// Handle incoming message from Iota
+    pub async fn handle_message(self: Arc<Self>, message: String) {
+        let cv = CommunicationValue::from_json(&message);
+
+        // Handle identification
+        if cv.is_type(CommunicationType::identification) && !self.is_identified().await {
+            self.handle_identification(cv).await;
+            return;
+        }
+
+        if !self.is_identified().await {
+            return;
+        }
+
+        // Handle ping
+        if cv.is_type(CommunicationType::ping) {
+            self.handle_ping(cv).await;
+            return;
+        }
+
+        // Handle forwarding to other Iotas or clients
+        let receiver_id = cv.get_receiver();
+        if !self.get_user_ids().await.contains(&receiver_id)
+            || cv.is_type(CommunicationType::message_other_iota)
+            || cv.is_type(CommunicationType::send_chat)
+        {
+            self.handle_forward_message(cv).await;
+            return;
+        }
+
+        // Handle GET_CHATS
+        if cv.is_type(CommunicationType::get_chats) {
+            self.handle_get_chats(cv).await;
+            return;
+        }
+
+        // Forward to client
+        self.forward_to_client(cv).await;
+    }
+
+    /// Handle identification message
+    async fn handle_identification(self: Arc<Self>, mut cv: CommunicationValue) {
+        // Parse Iota ID
+        let iota_id = match cv.get_data(DataTypes::iota_id) {
+            Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(&cv.get_id()).await;
+                    return;
+                }
+            },
+            None => {
+                self.send_error_response(&cv.get_id()).await;
+                return;
+            }
+        };
+
+        // Parse user IDs
+        let mut validated_user_ids = Vec::new();
+        if let Some(user_ids_str) = cv.get_data(DataTypes::user_ids) {
+            for id_str in user_ids_str.to_string().split(',') {
+                if id_str.is_empty() {
+                    continue;
+                }
+                if let Ok(user_id) = Uuid::parse_str(id_str.trim()) {
+                    let auth_iota_id = auth_connector::get_iota_id(user_id).await.unwrap();
+                    if auth_iota_id == iota_id {
+                        validated_user_ids.push(user_id);
+                    }
+                }
+            }
+        }
+
+        // Set identification data
+        {
+            let mut iota_id_guard = self.iota_id.write().await;
+            *iota_id_guard = iota_id;
+        }
+        {
+            let mut user_ids_guard = self.user_ids.write().await;
+            *user_ids_guard = validated_user_ids.clone();
+        }
+        {
+            let mut identified_guard = self.identified.write().await;
+            *identified_guard = true;
+        }
+
+        // Check for existing connection and close it
+        if rho_manager::contains_iota(iota_id).await {
+            if let Some(existing_rho) = rho_manager::get_rho_by_iota(iota_id).await {
+                existing_rho.close_iota_connection().await;
+            }
+        }
+
+        // Create RhoConnection
+        let rho_connection =
+            Arc::new(RhoConnection::new(self.clone(), validated_user_ids.clone()).await);
+
+        // Set up bidirectional reference
+        self.set_rho_connection(Arc::downgrade(&rho_connection))
+            .await;
+
+        // Add to manager
+        rho_manager::add_rho(rho_connection).await;
+
+        // Send response
+        let response = CommunicationValue::new(CommunicationType::identification_response)
+            .with_id(cv.get_id())
+            .add_data_str(DataTypes::accepted, validated_user_ids.len().to_string());
+
+        self.send_message(response).await;
+    }
+
+    /// Handle ping message
+    async fn handle_ping(&self, mut cv: CommunicationValue) {
+        // Update our ping if provided
+        if let Some(last_ping) = cv.get_data(DataTypes::last_ping) {
+            if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
+                let mut ping_guard = self.ping.write().await;
+                *ping_guard = ping_val;
+            }
+        }
+
+        // Collect client pings
+        let client_pings = if let Some(rho_conn) = self.get_rho_connection().await {
+            rho_conn.get_client_pings().await
+        } else {
+            HashMap::new()
+        };
+
+        // Send pong response
+        let pings = client_pings
+            .into_iter()
+            .map(|(k, v)| (k, JsonValue::String(v.to_string())))
+            .collect();
+        let response = CommunicationValue::new(CommunicationType::pong)
+            .with_id(cv.get_id())
+            .add_data(DataTypes::ping_clients, JsonValue::Object(pings));
+        self.send_message(response).await;
+    }
+
+    /// Handle message forwarding to other Iotas
+    async fn handle_forward_message(&self, cv: CommunicationValue) {
+        let receiver_id = cv.get_receiver();
+        // Validate sender if present
+        let sender_id = cv.get_sender();
+        if !self.get_user_ids().await.contains(&sender_id) {
+            self.send_error_response(&cv.get_id()).await;
+            return;
+        }
+
+        // Find target RhoConnection
+        if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id).await {
+            target_rho.message_to_iota(cv).await;
+        } else {
+            // Send error if target not found
+            let error = CommunicationValue::new(CommunicationType::error)
+                .with_id(cv.get_id())
+                .with_sender(cv.get_sender());
+            self.send_message(error).await;
+        }
+    }
+
+    /// Handle GET_CHATS message
+    async fn handle_get_chats(&self, mut cv: CommunicationValue) {
+        let receiver_id = cv.get_receiver();
+        let mut interested_ids: Vec<Uuid> = Vec::new();
+
+        // Process contacts and add call information
+        if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
+            // Parse contacts JSON array and enrich with call data
+            // This would need proper JSON parsing implementation
+            // For now, placeholder logic:
+
+            // Extract user IDs from contacts
+            // let contacts: Vec<ContactInfo> = parse_contacts(contacts_data);
+            // for contact in &contacts {
+            //     interested_ids.push(contact.user_id);
+            // }
+
+            // Get call invites for receiver
+            // let invites = CallManager::get_call_invites(receiver_id).await;
+
+            // Enrich contacts with call information
+            // let enriched_contacts = enrich_with_calls(contacts, invites);
+
+            // cv = cv.add_data(DataTypes::user_ids, enriched_contacts.into());
+        }
+
+        // Notify OmegaConnection about user states
+        OmegaConnection::user_states(receiver_id, interested_ids.clone());
+
+        // Set interested users in RhoConnection
+        if let Some(rho_conn) = self.get_rho_connection().await {
+            rho_conn.set_interested(receiver_id, interested_ids).await;
+        }
+
+        // Forward to client
+        self.forward_to_client(cv).await;
+    }
+
+    /// Forward message to client
+    async fn forward_to_client(&self, cv: CommunicationValue) {
+        if let Some(rho_conn) = self.get_rho_connection().await {
+            let updated_cv = cv.with_sender(self.get_iota_id().await);
+            let receiver_id = updated_cv.get_receiver();
+            rho_conn.message_iota_to_client(updated_cv).await;
+        }
+    }
+
+    /// Send error response
+    async fn send_error_response(&self, message_id: &Uuid) {
+        let error = CommunicationValue::new(CommunicationType::error).with_id(*message_id);
+        self.send_message(error).await;
+    }
+
+    /// Handle connection close
+    pub async fn handle_close(&self) {
+        if self.is_identified().await {
+            if let Some(rho_conn) = self.get_rho_connection().await {
+                rho_conn.close_iota_connection().await;
+            }
+        }
+    }
+}
+
+impl std::fmt::Debug for IotaConnection {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("IotaConnection")
+            .field("iota_id", &"[async]")
+            .field("identified", &"[async]")
+            .field("ping", &"[async]")
+            .finish()
+    }
+}

src/rho/mod.rs

@@ -0,0 +1,21 @@
+//! Rho module - Connection management between Iota and Client connections
+//!
+//! This module implements the Rho connection management system that maps
+//! single Iota connections to multiple Client connections, providing
+//! bidirectional communication capabilities.
+
+pub mod client_connection;
+pub mod iota_connection;
+pub mod rho_connection;
+pub mod rho_manager;
+
+// Re-export commonly used types for convenience
+pub use client_connection::ClientConnection;
+pub use iota_connection::IotaConnection;
+pub use rho_connection::RhoConnection;
+
+// Re-export key manager functions
+pub use rho_manager::{
+    add_rho, connection_count, contains_iota, get_all_connections, get_rho_by_iota,
+    get_rho_con_for_user, remove_rho,
+};

src/rho/rho_connection.rs

@@ -0,0 +1,251 @@
+use std::collections::{self, HashMap};
+use std::sync::{Arc, Weak};
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager};
+use crate::data::{
+    communication::{CommunicationType, CommunicationValue, DataTypes},
+    user::UserStatus,
+};
+use crate::omega::omega_connection::OmegaConnection;
+
+pub struct RhoConnection {
+    iota_connection: Arc<IotaConnection>,
+    user_ids: Vec<Uuid>,
+    client_connections: Arc<RwLock<Vec<Arc<ClientConnection>>>>,
+}
+
+impl RhoConnection {
+    /// Create a new RhoConnection
+    pub async fn new(iota_connection: Arc<IotaConnection>, user_ids: Vec<Uuid>) -> Self {
+        let rho_connection = Self {
+            iota_connection,
+            user_ids: user_ids.clone(),
+            client_connections: Arc::new(RwLock::new(Vec::new())),
+        };
+
+        // Notify OmegaConnection about the new Iota
+        OmegaConnection::connect_iota(rho_connection.get_iota_id().await, user_ids);
+
+        rho_connection
+    }
+
+    /// Get the Iota ID
+    pub async fn get_iota_id(&self) -> Uuid {
+        self.iota_connection.get_iota_id().await
+    }
+
+    /// Get the user IDs associated with this Rho connection
+    pub fn get_user_ids(&self) -> &Vec<Uuid> {
+        &self.user_ids
+    }
+
+    /// Get reference to the IotaConnection
+    pub fn get_iota_connection(&self) -> &Arc<IotaConnection> {
+        &self.iota_connection
+    }
+
+    /// Get all client connections
+    pub async fn get_client_connections(&self) -> Vec<Arc<ClientConnection>> {
+        let connections = self.client_connections.read().await;
+        connections.clone()
+    }
+
+    /// Get client connections for a specific user
+    pub async fn get_client_connections_for_user(
+        &self,
+        user_id: Uuid,
+    ) -> Vec<Arc<ClientConnection>> {
+        let connections = self.client_connections.read().await;
+        let mut collections = Vec::new();
+        for con in connections.iter() {
+            if con.get_user_id().await.unwrap() == user_id {
+                collections.push(con.clone());
+            }
+        }
+        collections
+    }
+
+    /// Add a client connection
+    pub async fn add_client_connection(&self, connection: Arc<ClientConnection>) {
+        // Notify Iota about new client
+        let notification = CommunicationValue::new(CommunicationType::client_connected)
+            .add_data_str(
+                DataTypes::user_id,
+                connection
+                    .get_user_id()
+                    .await
+                    .unwrap_or(Uuid::nil())
+                    .to_string(),
+            );
+
+        self.iota_connection.send_message(notification).await;
+
+        // Add to our list
+        {
+            let mut connections = self.client_connections.write().await;
+            connections.push(Arc::clone(&connection));
+        }
+
+        // Notify OmegaConnection
+        OmegaConnection::client_changed(
+            self.get_iota_id().await,
+            connection.get_user_id().await.unwrap_or(Uuid::nil()),
+            UserStatus::online,
+        );
+    }
+
+    /// Remove a client connection
+    pub async fn close_client_connection(&self, connection: Arc<ClientConnection>) {
+        {
+            let mut connections = self.client_connections.write().await;
+
+            let target_user_id = connection.get_user_id().await.unwrap();
+
+            connections.retain(|con| {
+                futures::executor::block_on(async {
+                    con.get_user_id().await.unwrap() != target_user_id
+                })
+            });
+
+            // Push the new connection
+            connections.push(Arc::clone(&connection));
+        }
+
+        // Notify OmegaConnection
+        OmegaConnection::client_changed(
+            self.get_iota_id().await,
+            connection.get_user_id().await.unwrap_or(Uuid::nil()),
+            UserStatus::user_offline,
+        );
+    }
+
+    /// Close the Iota connection and all associated client connections
+    pub async fn close_iota_connection(&self) {
+        // Close all client connections
+        let connections = self.get_client_connections().await;
+        for connection in connections {
+            connection.close().await;
+        }
+
+        // Remove from manager
+        rho_manager::remove_rho(self.get_iota_id().await).await;
+
+        // Notify OmegaConnection
+        OmegaConnection::close_iota(self.get_iota_id().await);
+    }
+
+    /// Send message from Iota to specific client by user ID
+    pub async fn message_iota_to_client_by_user(&self, user_id: Uuid, message: &str) {
+        let connections = self.client_connections.read().await;
+        for connection in connections.iter() {
+            if let Some(conn_user_id) = connection.get_user_id().await {
+                if conn_user_id == user_id {
+                    connection.send_message_str(message).await;
+                }
+            }
+        }
+    }
+
+    /// Send message from Iota to specific client
+    pub async fn message_iota_to_client(&self, cv: CommunicationValue) {
+        if let Some(receiver_id) = Some(cv.get_receiver()) {
+            let connections = self.client_connections.read().await;
+            for connection in connections.iter() {
+                if let Some(conn_user_id) = connection.get_user_id().await {
+                    if conn_user_id == receiver_id {
+                        connection.send_message(&cv).await;
+                    }
+                }
+            }
+        }
+    }
+
+    /// Send message to Iota as string
+    pub async fn message_to_iota_str(&self, message: &str) {
+        self.iota_connection.send_message_str(message).await;
+    }
+
+    /// Send message to Iota
+    pub async fn message_to_iota(&self, cv: CommunicationValue) {
+        self.iota_connection.send_message(cv).await;
+    }
+
+    /// Set interested users for a specific client
+    pub async fn set_interested(&self, user_id: Uuid, interested_ids: Vec<Uuid>) {
+        let connections = self.client_connections.read().await;
+        for connection in connections.iter() {
+            if let Some(conn_user_id) = connection.get_user_id().await {
+                if conn_user_id == user_id {
+                    connection
+                        .set_interested_users(interested_ids.clone())
+                        .await;
+                    break;
+                }
+            }
+        }
+    }
+
+    /// Check if clients are interested in a user
+    pub async fn are_they_interested(&self, user: &crate::data::user::User) {
+        let connections = self.client_connections.read().await;
+        for connection in connections.iter() {
+            connection.are_you_interested(user).await;
+        }
+    }
+
+    /// Get ping information for all clients
+    pub async fn get_client_pings(&self) -> HashMap<String, i64> {
+        let connections = self.client_connections.read().await;
+        let mut pings = HashMap::new();
+
+        for connection in connections.iter() {
+            if let Some(user_id) = connection.get_user_id().await {
+                pings.insert(user_id.to_string(), connection.get_ping().await);
+            }
+        }
+
+        pings
+    }
+
+    /// Check if this RhoConnection contains a specific user ID
+    pub fn contains_user(&self, user_id: &Uuid) -> bool {
+        self.user_ids.contains(user_id)
+    }
+
+    /// Get count of active client connections
+    pub async fn client_count(&self) -> usize {
+        let connections = self.client_connections.read().await;
+        connections.len()
+    }
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::sync::Mutex;
+
+    #[tokio::test]
+    async fn test_rho_connection_creation() {
+        let iota_id = Uuid::new_v4();
+        let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()];
+
+        // Mock session
+        // Create a mock WebSocket stream (this would fail in practice but shows the API)
+        // In real implementation, this would be a proper WebSocketStream
+        let mock_stream =
+            std::ptr::null_mut() as *mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>;
+        let mock_stream = unsafe { std::ptr::read(mock_stream) };
+        let iota_conn = IotaConnection::new_with_ids(
+            iota_id,
+            user_ids.clone(),
+            Arc::new(tokio::sync::Mutex::new(mock_stream)),
+        );
+
+        let rho_conn = RhoConnection::new(iota_conn, user_ids.clone()).await;
+
+        assert_eq!(rho_conn.get_iota_id().await, iota_id);
+        assert_eq!(rho_conn.get_user_ids(), &user_ids);
+        assert_eq!(rho_conn.client_count().await, 0);
+    }
+}

src/rho/rho_manager.rs

@@ -0,0 +1,93 @@
+use std::{
+    collections::HashMap,
+    sync::{Arc, LazyLock},
+};
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+use super::rho_connection::RhoConnection;
+
+// Static storage for RhoConnections, keyed by Iota ID
+pub static RHO_CONNECTIONS: LazyLock<Arc<RwLock<HashMap<Uuid, Arc<RhoConnection>>>>> =
+    LazyLock::new(|| Arc::new(RwLock::new(HashMap::new())));
+
+/// Get a RhoConnection for a specific user ID
+pub async fn get_rho_con_for_user(user_id: Uuid) -> Option<Arc<RhoConnection>> {
+    let connections = RHO_CONNECTIONS.read().await;
+    for rho_connection in connections.values() {
+        if rho_connection.get_user_ids().contains(&user_id) {
+            return Some(Arc::clone(rho_connection));
+        }
+    }
+    None
+}
+
+/// Check if an Iota ID exists in the connections
+pub async fn contains_iota(iota_id: Uuid) -> bool {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.contains_key(&iota_id)
+}
+
+/// Remove a RhoConnection by Iota ID
+pub async fn remove_rho(iota_id: Uuid) -> Option<Arc<RhoConnection>> {
+    let mut connections = RHO_CONNECTIONS.write().await;
+    connections.remove(&iota_id)
+}
+
+/// Add a RhoConnection to the manager
+pub async fn add_rho(rho_connection: Arc<RhoConnection>) {
+    let mut connections = RHO_CONNECTIONS.write().await;
+    let iota_id = rho_connection.get_iota_id().await;
+    connections.insert(iota_id, rho_connection);
+}
+
+/// Get a RhoConnection by Iota ID directly
+pub async fn get_rho_by_iota(iota_id: Uuid) -> Option<Arc<RhoConnection>> {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.get(&iota_id).map(Arc::clone)
+}
+
+/// Get all active RhoConnections
+pub async fn get_all_connections() -> Vec<Arc<RhoConnection>> {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.values().map(Arc::clone).collect()
+}
+
+/// Get the count of active connections
+pub async fn connection_count() -> usize {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.len()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::rho::iota_connection::IotaConnection;
+    use std::sync::Arc;
+    use tokio::sync::Mutex;
+
+    #[tokio::test]
+    async fn test_add_and_get_rho() {
+        // Clear any existing connections
+        {
+            let mut connections = RHO_CONNECTIONS.write().await;
+            connections.clear();
+        }
+
+        let iota_id = Uuid::new_v4();
+        let user_ids = vec![Uuid::new_v4(), Uuid::new_v4()];
+
+        // Skip this test due to WebSocket complexity - would require proper mock setup
+        return;
+
+        // This test would need proper WebSocket stream mocking:
+        // let mock_session = create_mock_websocket_stream();
+        // let iota_conn = IotaConnection::new_with_ids(iota_id, user_ids.clone(), mock_session);
+        // let rho_conn = Arc::new(RhoConnection::new(iota_conn, user_ids.clone()));
+
+        // Test assertions would go here:
+        // add_rho(Arc::clone(&rho_conn)).await;
+        // assert!(contains_iota(iota_id).await);
+        // etc.
+    }
+}