Sync

src/rho/client_connection.rs

@@ -0,0 +1,583 @@
+use crate::anonymous_clients::anonymous_manager;
+use crate::calls::{call_manager, call_util};
+use crate::omega::omega_connection::get_omega_connection;
+use crate::rho::connection::GeneralConnection;
+use crate::rho::{rho_connection::RhoConnection, rho_manager};
+use crate::util::logger::PrintType;
+use crate::{data::user::UserStatus, omega::omega_connection::OmegaConnection};
+use crate::{log_cv_in, log_cv_out, log_err, log_in, log_out};
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use tokio::sync::RwLock;
+use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
+use ttp_native::{Receiver, Sender};
+use uuid::Uuid;
+
+pub struct ClientConnection {
+    pub user_id: u64,
+
+    pub sender: Arc<Sender>,
+    pub receiver: Arc<Receiver>,
+    pub ping: Arc<RwLock<i64>>,
+    pub_key: Arc<RwLock<Option<Vec<u8>>>>,
+    pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>,
+    pub interested_users: Arc<RwLock<Vec<i64>>>,
+    is_open: Arc<RwLock<bool>>,
+}
+
+impl ClientConnection {
+    pub async fn from_general(general: Arc<GeneralConnection>, user_id: u64) -> Arc<Self> {
+        Arc::new(Self {
+            ping: Arc::new(RwLock::new(0)),
+            pub_key: Arc::new(RwLock::new(None)),
+            rho_connection: general.rho_connection.clone(),
+            interested_users: Arc::new(RwLock::new(Vec::new())),
+            is_open: Arc::new(RwLock::new(true)),
+            sender: general.sender.clone(),
+            receiver: general.receiver.clone(),
+            user_id: user_id,
+        })
+    }
+    pub fn start(self: Arc<Self>) {
+        let self_clone = self.clone();
+        tokio::spawn(async move {
+            while let Ok(cv) = self_clone.receiver.receive().await {
+                self_clone.clone().handle_message(cv).await;
+            }
+            self_clone.handle_close().await;
+        });
+
+        let self_clone2 = self.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(50)).await;
+            if self_clone2.get_rho_connection().await.is_none() {
+                self_clone2
+                    .send_error_response(0, CommunicationType::error)
+                    .await;
+            }
+        });
+    }
+
+    /// Get the user ID
+    pub async fn get_user_id(&self) -> u64 {
+        self.user_id
+    }
+
+    /// Get current ping
+    pub async fn get_ping(&self) -> i64 {
+        *self.ping.read().await
+    }
+
+    /// Get RhoConnection if available
+    pub async fn get_rho_connection(&self) -> Option<Arc<RhoConnection>> {
+        self.rho_connection.read().await.clone()
+    }
+
+    /// Send a CommunicationValue to the client
+    pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
+        if !*self.is_open.read().await {
+            log_out!(
+                self.user_id as i64,
+                PrintType::Client,
+                "Attempted to send message to a closed connection."
+            );
+            return;
+        }
+        if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) {
+            log_cv_out!(PrintType::Client, &cv);
+        }
+        let _ = self.sender.send(&cv).await;
+    }
+
+    /// Handle incoming message from client
+    pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
+        tokio::spawn(async move {
+            if cv.is_type(CommunicationType::ping) {
+                self.handle_ping(cv).await;
+                return;
+            }
+            log_cv_in!(PrintType::Client, cv);
+
+            // 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::call_token) {
+                self.handle_get_call(cv).await;
+                return;
+            }
+
+            if cv.is_type(CommunicationType::call_disconnect_user) {
+                self.handle_call_disconnect_user(cv).await;
+                return;
+            }
+
+            if cv.is_type(CommunicationType::call_timeout_user) {
+                self.handle_call_timeout_user(cv).await;
+                return;
+            }
+
+            if cv.is_type(CommunicationType::call_set_anonymous_joining) {
+                self.handle_call_set_anonymous_joining(cv).await;
+                return;
+            }
+            if cv.is_type(CommunicationType::get_user_data) {
+                if let Some(anonymous) = {
+                    if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
+                        anonymous_manager::get_anonymous_user(user_id as u64).await
+                    } else if let Some(username) = cv.get_data(DataTypes::username).as_str() {
+                        anonymous_manager::get_anonymous_user_by_name(username.to_string()).await
+                    } else {
+                        None
+                    }
+                } {
+                    let response = CommunicationValue::new(CommunicationType::get_user_data)
+                        .with_id(cv.get_id())
+                        .add_data(
+                            DataTypes::username,
+                            DataValue::Str(anonymous.get_user_name().await),
+                        )
+                        .add_data(
+                            DataTypes::user_id,
+                            DataValue::Number(anonymous.get_user_id() as i64),
+                        )
+                        .add_data(
+                            DataTypes::display,
+                            DataValue::Str(anonymous.get_display_name().await),
+                        )
+                        .add_data(
+                            DataTypes::avatar,
+                            DataValue::Str(anonymous.get_avatar().await),
+                        )
+                        .add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
+
+                    self.send_message(&response).await;
+
+                    return;
+                }
+            }
+
+            if cv.is_type(CommunicationType::change_user_data)
+                || cv.is_type(CommunicationType::read_notification)
+                || cv.is_type(CommunicationType::get_notifications)
+                || cv.is_type(CommunicationType::get_user_data)
+                || cv.is_type(CommunicationType::get_iota_data)
+                || cv.is_type(CommunicationType::delete_user)
+            {
+                let sender = self.get_user_id().await;
+                self.handle_omega_forward(cv.with_sender(sender as u64))
+                    .await;
+                return;
+            }
+            // Forward other messages to Iota
+            self.forward_to_iota(cv).await;
+        });
+    }
+    async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
+        let client_for_closure = self.clone();
+        tokio::spawn(async move {
+            let response_cv = get_omega_connection()
+                .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
+                .await;
+            if let Ok(response_cv) = response_cv {
+                client_for_closure.send_message(&response_cv).await;
+            }
+        });
+    }
+
+    /// Handle ping message
+    async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
+        // Update our ping if provided
+        if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
+            let current = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_millis();
+            let mut ping_guard = self.ping.write().await;
+            *ping_guard = current as i64 - last_ping;
+        }
+
+        // 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(DataTypes::ping_iota, DataValue::Number(iota_ping));
+
+        self.send_message(&response).await;
+    }
+
+    /// Handle client status change
+    async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
+        let user_id = self.get_user_id().await;
+        if let DataValue::Str(_status_str) = cv.get_data(DataTypes::user_state) {
+            let user_status = UserStatus::user_online;
+            if let Some(rho_conn) = self.get_rho_connection().await {
+                OmegaConnection::client_changed(
+                    rho_conn.get_iota_id().await as i64,
+                    user_id as i64,
+                    user_status,
+                )
+                .await;
+            }
+        }
+    }
+
+    /// Handle call invite
+    async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
+        let receiver_id: i64 = cv.get_data(DataTypes::receiver_id).as_number().unwrap_or(0);
+        if receiver_id == 0 {
+            self.send_error_response(cv.get_id(), CommunicationType::error_no_user_id)
+                .await;
+            return;
+        }
+
+        let call_id = match cv.get_data(DataTypes::call_id) {
+            DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
+                        .await;
+                    return;
+                }
+            },
+            _ => {
+                self.send_error_response(cv.get_id(), CommunicationType::error_no_call_id)
+                    .await;
+                return;
+            }
+        };
+
+        let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await;
+        if !invited {
+            self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
+                .await;
+            return;
+        }
+
+        // Find target RhoConnection
+        let target_rho = match rho_manager::get_rho_con_for_user(receiver_id).await {
+            Some(rho) => rho,
+            _ => {
+                self.send_error_response(cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        // Get sender user ID
+        let sender_id = self.get_user_id().await;
+
+        // Create and send call distribution message
+        let forward = CommunicationValue::new(CommunicationType::call_invite)
+            .with_receiver(receiver_id as u64)
+            .with_sender(sender_id as u64)
+            .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
+            .add_data(
+                DataTypes::receiver_id,
+                DataValue::Str(receiver_id.to_string()),
+            )
+            .add_data(DataTypes::sender_id, DataValue::Str(sender_id.to_string()));
+
+        target_rho.message_to_client(forward).await;
+
+        let response = CommunicationValue::new(CommunicationType::success).with_id(cv.get_id());
+        self.send_message(&response).await;
+    }
+
+    /// Handle get call request
+    async fn handle_get_call(self: Arc<Self>, cv: CommunicationValue) {
+        let user_id = self.get_user_id().await;
+
+        let call_id = match cv.get_data(DataTypes::call_id) {
+            DataValue::Str(id_str) => match Uuid::parse_str(id_str.as_str()) {
+                Ok(id) => id,
+                Err(_) => {
+                    self.send_error_response(cv.get_id(), CommunicationType::error)
+                        .await;
+                    return;
+                }
+            },
+            _ => {
+                self.send_error_response(cv.get_id(), CommunicationType::error)
+                    .await;
+                return;
+            }
+        };
+
+        if let Some(token) = call_manager::get_call_token(user_id, call_id).await {
+            let response = CommunicationValue::new(CommunicationType::call_token)
+                .with_id(cv.get_id())
+                .with_receiver(user_id as u64)
+                .add_data(DataTypes::call_token, DataValue::Str(token));
+            self.send_message(&response).await;
+        } else {
+            self.send_error_response(cv.get_id(), CommunicationType::error)
+                .await;
+            return;
+        }
+    }
+    async fn handle_call_timeout_user(self: Arc<Self>, cv: CommunicationValue) {
+        let call_id =
+            Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
+        let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
+        let untill = cv.get_data(DataTypes::untill).as_number().unwrap_or(0);
+
+        let call = call_manager::get_call(call_id).await;
+        if let Some(call) = call {
+            if call
+                .get_caller(self.get_user_id().await)
+                .await
+                .unwrap()
+                .has_admin()
+            {
+                let _ = call_util::remove_participant(call_id, user_id as u64).await;
+                call.get_caller(user_id as u64)
+                    .await
+                    .unwrap()
+                    .set_timeout(untill)
+                    .await;
+            }
+        }
+    }
+    async fn handle_call_disconnect_user(self: Arc<Self>, cv: CommunicationValue) {
+        let call_id =
+            Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
+        let user_id = cv.get_data(DataTypes::user_id).as_number().unwrap_or(0);
+
+        let call = call_manager::get_call(call_id).await;
+        if let Some(call) = call {
+            if call
+                .get_caller(self.get_user_id().await)
+                .await
+                .unwrap()
+                .has_admin()
+            {
+                call.remove_caller(user_id as u64).await;
+            }
+        }
+    }
+    async fn handle_call_set_anonymous_joining(self: Arc<Self>, cv: CommunicationValue) {
+        let call_id =
+            Uuid::from_str(cv.get_data(DataTypes::call_id).as_str().unwrap_or("")).unwrap();
+        let enable = cv.get_data(DataTypes::enabled).as_bool().unwrap_or(true);
+
+        let call = call_manager::get_call(call_id).await;
+
+        let mut short_link = None;
+        if let Some(call) = call {
+            if call
+                .get_caller(self.get_user_id().await)
+                .await
+                .unwrap()
+                .has_admin()
+            {
+                call.set_anonymous_joining(enable).await;
+            }
+            short_link = call.get_short_link().await;
+        }
+        let mut response_cv =
+            CommunicationValue::new(CommunicationType::call_set_anonymous_joining)
+                .with_id(cv.get_id())
+                .add_data(DataTypes::call_id, DataValue::Str(call_id.to_string()))
+                .add_data(DataTypes::enabled, DataValue::Bool(enable));
+        if let Some(short_link) = short_link {
+            response_cv = response_cv.add_data(DataTypes::link, DataValue::Str(short_link));
+        }
+        self.send_message(&response_cv).await;
+    }
+
+    /// Forward message to Iota
+    async fn forward_to_iota(self: Arc<Self>, cv: CommunicationValue) {
+        let sender_user_id = self.get_user_id().await;
+        let msg_id = cv.get_id();
+        let msg_type = cv.get_type();
+
+        log_in!(
+            sender_user_id as i64,
+            PrintType::Client,
+            "Forwarding client->iota: sender={} type={:?} id={} receiver={}",
+            sender_user_id,
+            msg_type,
+            msg_id,
+            cv.get_receiver()
+        );
+
+        if cv.is_type(CommunicationType::add_conversation)
+            && cv
+                .get_data(DataTypes::chat_partner_id)
+                .as_number()
+                .is_none()
+        {
+            let chat_partner_name = cv
+                .get_data(DataTypes::chat_partner_name)
+                .as_str()
+                .unwrap_or("")
+                .to_string();
+
+            if anonymous_manager::get_anonymous_user_by_name(chat_partner_name.to_string())
+                .await
+                .is_some()
+            {
+                self.send_error_response(cv.get_id(), CommunicationType::error_anonymous)
+                    .await;
+                return;
+            }
+
+            let load_uuid_response = get_omega_connection()
+                .await_response(
+                    &CommunicationValue::new(CommunicationType::get_user_data)
+                        .with_id(cv.clone().get_id())
+                        .add_data(
+                            DataTypes::username,
+                            DataValue::Str(chat_partner_name.clone()),
+                        ),
+                    Some(Duration::from_secs(20)),
+                )
+                .await;
+            let chat_partner_id = {
+                if let Ok(load_uuid_response) = load_uuid_response {
+                    load_uuid_response.get_data(DataTypes::user_id).clone()
+                } else {
+                    DataValue::Null
+                }
+            };
+
+            if let Some(rho_conn) = self.get_rho_connection().await {
+                let iota_id = rho_conn.get_iota_id().await;
+                log_in!(
+                    sender_user_id as i64,
+                    PrintType::Client,
+                    "Resolved rho for add_conversation: sender={} -> iota_id={} id={}",
+                    sender_user_id,
+                    iota_id,
+                    msg_id
+                );
+
+                let updated_cv = cv
+                    .with_sender(sender_user_id as u64)
+                    .add_data(DataTypes::chat_partner_id, chat_partner_id);
+                rho_conn.message_to_iota(updated_cv).await;
+            } else {
+                log_err!(
+                    sender_user_id as i64,
+                    PrintType::Client,
+                    "No rho/iota mapping found for add_conversation sender={} type={:?} id={}",
+                    sender_user_id,
+                    msg_type,
+                    msg_id
+                );
+            }
+            return;
+        }
+
+        if let Some(rho_conn) = self.get_rho_connection().await {
+            let iota_id = rho_conn.get_iota_id().await;
+            log_in!(
+                sender_user_id as i64,
+                PrintType::Client,
+                "Resolved rho for forward: sender={} -> iota_id={} type={:?} id={}",
+                sender_user_id,
+                iota_id,
+                msg_type,
+                msg_id
+            );
+
+            let updated_cv = cv.with_sender(sender_user_id as u64);
+            rho_conn.message_to_iota(updated_cv).await;
+        } else {
+            log_err!(
+                sender_user_id as i64,
+                PrintType::Client,
+                "No rho/iota mapping found for sender={} type={:?} id={}",
+                sender_user_id,
+                msg_type,
+                msg_id
+            );
+            self.send_error_response(msg_id, CommunicationType::error)
+                .await;
+        }
+    }
+
+    /// Send error response
+    async fn send_error_response(self: Arc<Self>, message_id: u32, 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 is_open_guard = self.is_open.write().await;
+        if !*is_open_guard {
+            return;
+        }
+        *is_open_guard = false;
+
+        let _ = self.sender.close();
+    }
+
+    /// Set interested users list
+    pub async fn set_interested_users(self: Arc<Self>, interested_ids: Vec<i64>) {
+        let mut interested_guard = self.interested_users.write().await;
+        *interested_guard = interested_ids;
+    }
+    #[allow(dead_code)]
+    pub async fn get_interested_users(self: Arc<Self>) -> Vec<i64> {
+        let interested_guard = self.interested_users.read().await;
+        interested_guard.clone()
+    }
+
+    /// Check if interested in a user and send notification
+    #[allow(dead_code)]
+    pub async fn are_you_interested(self: Arc<Self>, user_id: i64) {
+        let interested_guard = self.clone().get_interested_users().await;
+        if interested_guard.contains(&user_id) {
+            let notification = CommunicationValue::new(CommunicationType::client_changed)
+                .add_data(DataTypes::user_id, DataValue::Str(user_id.to_string()))
+                .add_data(DataTypes::user_state, DataValue::Str("online".to_string()));
+
+            self.send_message(&notification).await;
+        }
+    }
+
+    /// Handle connection close
+    pub async fn handle_close(&self) {
+        let user_id = self.get_user_id().await;
+        if let Some(rho_conn) = rho_manager::get_rho_con_for_user(user_id as i64).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 {
+            sender: Arc::clone(&self.sender),
+            receiver: Arc::clone(&self.receiver),
+            user_id: self.user_id,
+            ping: Arc::clone(&self.ping),
+            pub_key: Arc::clone(&self.pub_key),
+            rho_connection: Arc::clone(&self.rho_connection),
+            interested_users: Arc::clone(&self.interested_users),
+            is_open: Arc::clone(&self.is_open),
+        }
+    }
+}

src/rho/connection.rs

@@ -0,0 +1,358 @@
+use rand::{Rng, distributions::Alphanumeric};
+use std::{sync::Arc, time::Duration};
+use tokio::sync::RwLock;
+use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
+use ttp_native::{Receiver, Sender};
+
+use crate::{
+    anonymous_clients::anonymous_client_connection::AnonymousClientConnection,
+    get_private_key, get_public_key, log_cv_in, log_cv_out, log_err, log_in, log_out,
+    omega::omega_connection::get_omega_connection,
+    rho::{
+        client_connection::ClientConnection, iota_connection::IotaConnection,
+        rho_connection::RhoConnection, rho_manager,
+    },
+    util::{
+        crypto_helper::{load_public_key, public_key_to_base64},
+        crypto_util::{DataFormat, SecurePayload},
+        logger::PrintType,
+    },
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[allow(dead_code)]
+pub enum ConnectionKind {
+    Client,
+    Iota,
+    AnonymousClient,
+    Phi,
+}
+
+pub struct GeneralConnection {
+    pub sender: Arc<Sender>,
+    pub receiver: Arc<Receiver>,
+
+    identified: Arc<RwLock<bool>>,
+    challenged: Arc<RwLock<bool>>,
+    challenge: Arc<RwLock<String>>,
+
+    connection_kind: Arc<RwLock<Option<ConnectionKind>>>,
+    pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>,
+    id: Arc<RwLock<u64>>,
+
+    pub_key: Arc<RwLock<Option<Vec<u8>>>>,
+}
+impl GeneralConnection {
+    pub fn new(sender: Sender, receiver: Receiver) -> Arc<Self> {
+        Arc::new(Self {
+            sender: Arc::new(sender),
+            receiver: Arc::new(receiver),
+            identified: Arc::new(RwLock::new(false)),
+            challenged: Arc::new(RwLock::new(false)),
+            challenge: Arc::new(RwLock::new(String::new())),
+            connection_kind: Arc::new(RwLock::new(None)),
+            rho_connection: Arc::new(RwLock::new(None)),
+            id: Arc::new(RwLock::new(0)),
+            pub_key: Arc::new(RwLock::new(None)),
+        })
+    }
+}
+impl GeneralConnection {
+    pub async fn handle(self: Arc<Self>) {
+        log_in!(0, PrintType::General, "General connection handler started");
+
+        loop {
+            let cv = match self.receiver.receive().await {
+                Ok(v) => v,
+                Err(_) => {
+                    break;
+                }
+            };
+
+            log_cv_in!(cv);
+
+            if !*self.identified.read().await {
+                self.handle_identification(cv).await;
+                continue;
+            }
+
+            if !*self.challenged.read().await {
+                self.handle_challenge_response(cv).await;
+            }
+
+            if *self.challenged.read().await {
+                let self_clone = self.clone();
+                tokio::spawn(async move {
+                    self_clone.migrate().await;
+                });
+                break;
+            }
+        }
+
+        log_out!(0, PrintType::General, "General connection handler stopped");
+    }
+    async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) {
+        if !cv.is_type(CommunicationType::identification) {
+            return;
+        }
+
+        if let DataValue::Number(iota_id) = cv.get_data(DataTypes::iota_id) {
+            *self.id.write().await = *iota_id as u64;
+            *self.connection_kind.write().await = Some(ConnectionKind::Iota);
+
+            let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data)
+                .add_data(DataTypes::iota_id, DataValue::Number(*iota_id));
+
+            let response_cv = get_omega_connection()
+                .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
+                .await;
+
+            let response_cv = match response_cv {
+                Ok(r) => r,
+                Err(_) => {
+                    return;
+                }
+            };
+
+            let base64_pub = response_cv
+                .get_data(DataTypes::public_key)
+                .as_str()
+                .unwrap_or("");
+
+            let pub_key = match load_public_key(base64_pub) {
+                Some(pk) => pk,
+                None => {
+                    return;
+                }
+            };
+
+            *self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
+
+            let challenge: String = rand::thread_rng()
+                .sample_iter(&Alphanumeric)
+                .take(32)
+                .map(char::from)
+                .collect();
+
+            *self.challenge.write().await = challenge.clone();
+            *self.identified.write().await = true;
+
+            let encrypted_challenge =
+                SecurePayload::new(challenge.as_bytes(), DataFormat::Raw, get_private_key())
+                    .unwrap()
+                    .encrypt_x448(pub_key)
+                    .unwrap()
+                    .export(DataFormat::Base64);
+
+            let response = CommunicationValue::new(CommunicationType::challenge)
+                .with_id(cv.get_id())
+                .add_data(
+                    DataTypes::public_key,
+                    DataValue::Str(public_key_to_base64(&get_public_key())),
+                )
+                .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
+
+            log_cv_out!(response);
+            let _ = self.sender.send(&response).await;
+        } else if let DataValue::Number(user_id) = cv.get_data(DataTypes::user_id) {
+            *self.id.write().await = *user_id as u64;
+            *self.connection_kind.write().await = Some(ConnectionKind::Client);
+
+            let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_user_data)
+                .add_data(DataTypes::user_id, DataValue::Number(*user_id));
+
+            let response_cv = get_omega_connection()
+                .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
+                .await;
+
+            let response_cv = match response_cv {
+                Ok(r) => r,
+                Err(_) => {
+                    return;
+                }
+            };
+
+            let base64_pub = response_cv
+                .get_data(DataTypes::public_key)
+                .as_str()
+                .unwrap_or("");
+
+            let pub_key = match load_public_key(base64_pub) {
+                Some(pk) => pk,
+                None => {
+                    return;
+                }
+            };
+
+            *self.pub_key.write().await = Some(pub_key.as_bytes().to_vec());
+
+            let challenge: String = rand::thread_rng()
+                .sample_iter(&Alphanumeric)
+                .take(32)
+                .map(char::from)
+                .collect();
+
+            *self.challenge.write().await = challenge.clone();
+            *self.identified.write().await = true;
+
+            let encrypted_challenge =
+                SecurePayload::new(challenge.as_bytes(), DataFormat::Raw, get_private_key())
+                    .unwrap()
+                    .encrypt_x448(pub_key)
+                    .unwrap()
+                    .export(DataFormat::Base64);
+
+            let response = CommunicationValue::new(CommunicationType::challenge)
+                .with_id(cv.get_id())
+                .add_data(
+                    DataTypes::public_key,
+                    DataValue::Str(public_key_to_base64(&get_public_key())),
+                )
+                .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
+
+            log_cv_out!(response);
+            let _ = self.sender.send(&response).await;
+        }
+    }
+    async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) {
+        let id = *self.id.read().await as i64;
+
+        if !cv.is_type(CommunicationType::challenge_response) {
+            return;
+        }
+
+        if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) {
+            let expected = self.challenge.read().await.clone();
+
+            if *response == expected {
+                *self.challenged.write().await = true;
+
+                let response = CommunicationValue::new(CommunicationType::identification_response)
+                    .with_id(cv.get_id())
+                    .add_data(DataTypes::accepted, DataValue::Bool(true));
+
+                log_cv_out!(response);
+                if let Err(_) = self.sender.send(&response).await {
+                    return;
+                }
+            } else {
+                log_err!(
+                    id,
+                    PrintType::Iota,
+                    "Challenge response mismatch expected={} actual={}",
+                    expected,
+                    response
+                );
+            }
+        } else {
+            log_err!(
+                id,
+                PrintType::Iota,
+                "Challenge response missing challenge payload"
+            );
+        }
+    }
+
+    async fn migrate(self: &Arc<Self>) -> bool {
+        let kind = match *self.connection_kind.read().await {
+            Some(kind) => kind,
+            None => {
+                return false;
+            }
+        };
+        let id = *self.id.read().await;
+
+        match kind {
+            ConnectionKind::Client => {
+                let notify = CommunicationValue::new(CommunicationType::user_connected)
+                    .add_data(DataTypes::user_id, DataValue::Number(id as i64));
+                get_omega_connection().send_message(&notify).await;
+
+                let user_id = id as i64;
+
+                let client = ClientConnection::from_general(self.clone(), id).await;
+
+                let mut rho = rho_manager::get_rho_con_for_user(user_id).await;
+
+                if rho.is_none() {
+                    let get_user_msg = CommunicationValue::new(CommunicationType::get_user_data)
+                        .add_data(DataTypes::user_id, DataValue::Number(user_id));
+
+                    if let Ok(user_data_cv) = get_omega_connection()
+                        .await_response(&get_user_msg, Some(Duration::from_secs(20)))
+                        .await
+                    {
+                        if let DataValue::Number(iota_id) =
+                            user_data_cv.get_data(DataTypes::iota_id)
+                        {
+                            if let Some(bound_rho) =
+                                rho_manager::bind_user_to_iota(user_id, *iota_id).await
+                            {
+                                bound_rho.bind_user_id(user_id).await;
+                                rho = Some(bound_rho);
+                            }
+                        }
+                    }
+                }
+
+                *self.rho_connection.write().await = rho.clone();
+
+                if let Some(rho_conn) = rho {
+                    rho_conn.bind_user_id(user_id).await;
+                    rho_conn.add_client_connection(client.clone()).await;
+                } else {
+                    log_err!(
+                        user_id,
+                        PrintType::Client,
+                        "No RhoConnection found for user {}, client not attached to iota",
+                        id
+                    );
+                }
+
+                client.start();
+            }
+            ConnectionKind::Iota => {
+                let notify = CommunicationValue::new(CommunicationType::iota_connected)
+                    .add_data(DataTypes::iota_id, DataValue::Number(id as i64));
+                get_omega_connection().send_message(&notify).await;
+
+                let iota = IotaConnection::from_general(self.clone(), id).await;
+
+                let rho = Arc::new(RhoConnection::new(iota.clone(), Vec::new()).await);
+
+                iota.set_rho_connection(rho.clone()).await;
+
+                rho_manager::add_rho(rho).await;
+
+                let get_iota_msg = CommunicationValue::new(CommunicationType::get_iota_data)
+                    .add_data(DataTypes::iota_id, DataValue::Number(id as i64));
+
+                if let Ok(iota_data_cv) = get_omega_connection()
+                    .await_response(&get_iota_msg, Some(Duration::from_secs(20)))
+                    .await
+                {
+                    if let DataValue::Array(users) = iota_data_cv.get_data(DataTypes::user_ids) {
+                        let mut user_ids: Vec<u64> = Vec::new();
+                        for value in users {
+                            if let DataValue::Number(user_id) = value {
+                                user_ids.push(*user_id as u64);
+                            }
+                        }
+                        iota.set_user_ids(user_ids).await;
+                    }
+                }
+
+                iota.clone().start();
+            }
+            ConnectionKind::AnonymousClient => {
+                let client = AnonymousClientConnection::from_general(self.clone(), id).await;
+                client.start();
+            }
+            ConnectionKind::Phi => {
+                let phi = ClientConnection::from_general(self.clone(), id).await;
+                phi.start();
+            }
+        }
+        true
+    }
+}

src/rho/iota_connection.rs

@@ -0,0 +1,461 @@
+use crate::calls::call_group::CallGroup;
+use crate::calls::call_manager;
+use crate::log_cv_in;
+use crate::log_cv_out;
+use crate::log_err;
+use crate::log_in;
+use crate::omega::omega_connection::get_omega_connection;
+use crate::rho::connection::GeneralConnection;
+use crate::util::logger::PrintType;
+use dashmap::DashMap;
+use std::collections::BTreeMap;
+use std::{collections::HashMap, sync::Arc, time::Duration};
+use tokio::sync::RwLock;
+use tokio::sync::mpsc;
+use ttp_core::CommunicationType;
+use ttp_core::CommunicationValue;
+use ttp_core::DataTypes;
+use ttp_core::DataValue;
+use ttp_native::Receiver;
+use ttp_native::Sender;
+use x448::PublicKey;
+
+use super::{rho_connection::RhoConnection, rho_manager};
+use crate::omega::omega_connection::OmegaConnection;
+
+#[allow(dead_code)]
+pub struct IotaConnection {
+    pub iota_id: u64,
+    pub sender: Arc<Sender>,
+    pub receiver: Arc<Receiver>,
+    pub user_ids: Arc<RwLock<Vec<u64>>>,
+    pub ping: Arc<RwLock<i64>>,
+    pub_key: Arc<RwLock<Option<Vec<u8>>>>,
+    pub waiting_tasks:
+        DashMap<u32, Box<dyn Fn(Arc<IotaConnection>, CommunicationValue) -> bool + Send + Sync>>,
+    pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>,
+}
+
+impl IotaConnection {
+    pub async fn from_general(general: Arc<GeneralConnection>, iota_id: u64) -> Arc<Self> {
+        Arc::new(Self {
+            ping: Arc::new(RwLock::new(0)),
+            pub_key: Arc::new(RwLock::new(None)),
+            rho_connection: general.rho_connection.clone(),
+            user_ids: Arc::new(RwLock::new(Vec::new())),
+            sender: general.sender.clone(),
+            receiver: general.receiver.clone(),
+            iota_id: iota_id,
+            waiting_tasks: DashMap::new(),
+        })
+    }
+    pub fn start(self: Arc<Self>) {
+        let self_clone = self.clone();
+        tokio::spawn(async move {
+            loop {
+                match self_clone.receiver.receive().await {
+                    Ok(cv) => {
+                        self_clone.clone().handle_message(cv).await;
+                    }
+                    Err(_) => {
+                        break;
+                    }
+                }
+            }
+            self_clone.handle_close().await;
+        });
+    }
+
+    /// Get the Iota ID
+    pub async fn get_iota_id(&self) -> u64 {
+        self.iota_id
+    }
+
+    #[allow(dead_code)]
+    pub async fn get_public_key(&self) -> Option<PublicKey> {
+        if let Some(public_key) = self.pub_key.read().await.clone() {
+            PublicKey::from_bytes(&public_key)
+        } else {
+            None
+        }
+    }
+
+    /// Get the user IDs
+    pub async fn get_user_ids(&self) -> Vec<u64> {
+        self.user_ids.read().await.clone()
+    }
+
+    /// Replace all users linked to this iota and synchronize the attached rho mapping.
+    pub async fn set_user_ids(&self, user_ids: Vec<u64>) {
+        {
+            let mut guard = self.user_ids.write().await;
+            *guard = user_ids.clone();
+        }
+
+        if let Some(rho_conn) = self.get_rho_connection().await {
+            let user_ids_i64: Vec<i64> = user_ids.into_iter().map(|u| u as i64).collect();
+            rho_conn.set_user_ids(user_ids_i64).await;
+        }
+    }
+
+    pub async fn add_user_id(&self, user_id: u64) {
+        let mut should_sync = false;
+        {
+            let mut guard = self.user_ids.write().await;
+            if !guard.contains(&user_id) {
+                guard.push(user_id);
+                should_sync = true;
+            }
+        }
+
+        if should_sync {
+            if let Some(rho_conn) = self.get_rho_connection().await {
+                rho_conn.add_user_id(user_id as i64).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: Arc<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() {
+            Some(weak_ref.clone())
+        } else {
+            None
+        }
+    }
+
+    /// Send a CommunicationValue to the Iota
+    pub async fn send_message(&self, cv: &CommunicationValue) {
+        if !cv.is_type(CommunicationType::pong) {
+            log_cv_out!(PrintType::Iota, cv);
+        }
+        if let Err(e) = self.sender.send(&cv).await {
+            log_err!(
+                self.iota_id as i64,
+                PrintType::Iota,
+                "Failed to send message: {:?}",
+                e
+            );
+        }
+    }
+
+    /// Handle incoming message from Iota
+    pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
+        let msg_id = cv.get_id();
+        if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) {
+            if (task)(self.clone(), cv.clone()) {
+                return;
+            }
+        }
+
+        // Handle ping
+        if cv.is_type(CommunicationType::ping) || cv.is_type(CommunicationType::pong) {
+            self.handle_ping(cv).await;
+            return;
+        }
+
+        log_cv_in!(PrintType::Iota, cv);
+
+        // Handle GET_CHATS
+        if cv.is_type(CommunicationType::get_chats) {
+            self.handle_get_chats(cv).await;
+            return;
+        }
+
+        // Handle forwarding to other Iotas or clients
+        let receiver_id = cv.get_receiver();
+        if (receiver_id != 0 && !self.get_user_ids().await.contains(&(receiver_id as u64)))
+            || cv.is_type(CommunicationType::message_other_iota)
+            || cv.is_type(CommunicationType::send_chat)
+        {
+            self.handle_forward_message(cv).await;
+            return;
+        }
+
+        if cv.is_type(CommunicationType::change_iota_data)
+            || cv.is_type(CommunicationType::push_notification)
+            || cv.is_type(CommunicationType::get_user_data)
+            || cv.is_type(CommunicationType::get_iota_data)
+            || cv.is_type(CommunicationType::get_register)
+            || cv.is_type(CommunicationType::complete_register_user)
+            || cv.is_type(CommunicationType::delete_iota)
+        {
+            let sender = self.get_iota_id().await;
+            self.handle_omega_forward(cv.with_sender(sender as u64))
+                .await;
+            return;
+        }
+        self.forward_to_client(cv).await;
+    }
+
+    #[allow(dead_code)]
+    async fn send_error_response(&self, message_id: u32, error_type: CommunicationType) {
+        let error = CommunicationValue::new(error_type).with_id(message_id);
+        self.send_message(&error).await;
+    }
+
+    #[allow(dead_code)]
+    async fn close(&self) {
+        let _ = self.sender.close();
+    }
+
+    async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
+        let iota_for_closure = self.clone();
+        tokio::spawn(async move {
+            let response_cv = get_omega_connection()
+                .await_response(&cv.with_sender(self.iota_id), Some(Duration::from_secs(20)))
+                .await;
+            if let Ok(response_cv) = response_cv {
+                iota_for_closure.send_message(&response_cv).await;
+            }
+        });
+    }
+    /// Handle ping message
+    async fn handle_ping(&self, cv: CommunicationValue) {
+        if let DataValue::Number(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;
+            }
+        }
+
+        let client_pings = if let Some(rho_conn) = self.get_rho_connection().await {
+            rho_conn.get_client_pings().await
+        } else {
+            HashMap::new()
+        };
+
+        let pings: Vec<(DataTypes, DataValue)> = client_pings
+            .into_iter()
+            .map(|(k, v)| (DataTypes::parse(k), DataValue::Number(v)))
+            .collect();
+        let response = CommunicationValue::new(CommunicationType::pong)
+            .with_id(cv.get_id())
+            .add_data(DataTypes::ping_clients, DataValue::Container(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();
+        let sender_id = cv.get_sender();
+        let my_user_ids = self.get_user_ids().await;
+
+        log_in!(
+            self.iota_id as i64,
+            PrintType::Iota,
+            "Authority check: sender_id={} receiver_id={} iota_user_ids={:?} msg_type={:?} msg_id={}",
+            sender_id,
+            receiver_id,
+            my_user_ids,
+            cv.get_type(),
+            cv.get_id()
+        );
+
+        if my_user_ids.contains(&(sender_id as u64)) {
+            if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await {
+                target_rho.message_to_iota(cv).await;
+            } else {
+                let error = CommunicationValue::new(CommunicationType::error_no_iota)
+                    .with_id(cv.get_id())
+                    .with_sender(cv.get_sender());
+                self.send_message(&error).await;
+            }
+        } else {
+            log_err!(
+                self.iota_id as i64,
+                PrintType::Iota,
+                "Rejected client->iota forward: sender_id={} is not authorized for this iota. Known users={:?}",
+                sender_id,
+                my_user_ids
+            );
+
+            self.send_message(
+                &CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data(
+                    DataTypes::error_type,
+                    DataValue::Str(
+                        "You are sending to another User without authority.".to_string(),
+                    ),
+                ),
+            )
+            .await;
+        }
+    }
+
+    /// Handle GET_CHATS message
+    async fn handle_get_chats(&self, cv: CommunicationValue) {
+        let receiver_id = cv.get_receiver();
+        let mut interested_ids: Vec<i64> = Vec::new();
+
+        // ============================
+        // Load Calls
+        // ============================
+        let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(receiver_id).await;
+
+        let mut invites: HashMap<i64, Vec<DataValue>> = HashMap::new();
+        let empty = calls.is_empty();
+
+        for call in calls {
+            for inviter in call.members.read().await.iter() {
+                let call_self = call.get_caller(receiver_id).await.unwrap();
+
+                let inviter_id = inviter.user_id;
+                let timeout = *call_self.timeout.read().await;
+                let admin = call_self.has_admin();
+
+                // Build call container
+                let mut call_map: BTreeMap<DataTypes, DataValue> = BTreeMap::new();
+
+                call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string()));
+
+                if timeout > 0 {
+                    call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64));
+                }
+
+                if admin {
+                    call_map.insert(DataTypes::has_admin, DataValue::Bool(true));
+                }
+
+                let call_container = DataValue::container_from_map(&call_map);
+
+                invites
+                    .entry(inviter_id as i64)
+                    .or_insert_with(Vec::new)
+                    .push(call_container);
+            }
+        }
+
+        // ============================
+        // Enrich Contacts
+        // ============================
+        let enriched_contacts = if empty {
+            match cv.get_data(DataTypes::user_ids) {
+                DataValue::Array(arr) => DataValue::Array(arr.clone()),
+                _ => DataValue::Array(vec![]),
+            }
+        } else {
+            let mut enriched: Vec<DataValue> = Vec::new();
+
+            if let DataValue::Array(users) = cv.get_data(DataTypes::user_ids) {
+                for user_val in users {
+                    if let DataValue::Container(entries) = user_val {
+                        let mut user_map: BTreeMap<DataTypes, DataValue> =
+                            entries.iter().cloned().collect();
+
+                        // extract user_id
+                        if let Some(DataValue::Number(user_id)) = user_map.get(&DataTypes::user_id)
+                        {
+                            interested_ids.push(*user_id);
+
+                            // attach calls if exists
+                            if let Some(call_list) = invites.get(user_id) {
+                                user_map
+                                    .insert(DataTypes::calls, DataValue::Array(call_list.clone()));
+                            }
+                        }
+
+                        enriched.push(DataValue::container_from_map(&user_map));
+                    }
+                }
+            }
+
+            DataValue::Array(enriched)
+        };
+
+        // ============================
+        // Notify Omega
+        // ============================
+        OmegaConnection::user_states(receiver_id as i64, interested_ids.clone()).await;
+
+        // ============================
+        // Notify Rho
+        // ============================
+        if let Some(rho_conn) = self.get_rho_connection().await {
+            rho_conn
+                .set_interested(receiver_id as i64, interested_ids)
+                .await;
+        }
+
+        // ============================
+        // Forward to client
+        // ============================
+        self.forward_to_client(cv.add_data(DataTypes::user_ids, enriched_contacts))
+            .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);
+            rho_conn.message_to_client(updated_cv).await;
+        } else {
+        }
+    }
+
+    pub async fn handle_close(&self) {
+        if let Some(rho_conn) = self.get_rho_connection().await {
+            rho_conn.close_iota_connection().await;
+        }
+    }
+
+    #[allow(dead_code)]
+    pub async fn await_response(
+        self: Arc<IotaConnection>,
+        cv: &CommunicationValue,
+        timeout_duration: Option<Duration>,
+    ) -> Result<CommunicationValue, String> {
+        let (tx, mut rx) = mpsc::channel(1);
+        let msg_id = cv.get_id();
+
+        let task_tx = tx.clone();
+        self.waiting_tasks.insert(
+            msg_id,
+            Box::new(move |_, response_cv| {
+                let inner_tx = task_tx.clone();
+                tokio::spawn(async move {
+                    let _ = inner_tx.send(response_cv).await;
+                });
+                true
+            }),
+        );
+
+        self.send_message(cv).await;
+
+        let timeout = timeout_duration.unwrap_or(Duration::from_secs(10));
+
+        match tokio::time::timeout(timeout, rx.recv()).await {
+            Ok(Some(response_cv)) => Ok(response_cv),
+            Ok(_) => Err("Failed to receive response, channel was closed.".to_string()),
+            Err(_) => {
+                self.waiting_tasks.remove(&msg_id);
+                Err(format!(
+                    "Request timed out after {} seconds.",
+                    timeout.as_secs()
+                ))
+            }
+        }
+    }
+}
+
+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,6 @@
+pub mod client_connection;
+pub mod connection;
+pub mod iota_connection;
+pub mod rho_connection;
+pub mod rho_manager;
+pub mod server;

src/rho/rho_connection.rs

@@ -0,0 +1,199 @@
+use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager};
+
+use crate::data::user::UserStatus;
+use crate::omega::omega_connection::OmegaConnection;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
+
+pub struct RhoConnection {
+    iota_connection: Arc<IotaConnection>,
+    user_ids: Arc<RwLock<Vec<i64>>>,
+    client_connections: Arc<RwLock<Vec<Arc<ClientConnection>>>>,
+}
+
+impl RhoConnection {
+    /// Create a new RhoConnection
+    pub async fn new(iota_connection: Arc<IotaConnection>, user_ids: Vec<i64>) -> Self {
+        let rho_connection = Self {
+            iota_connection,
+            user_ids: Arc::new(RwLock::new(user_ids.clone())),
+            client_connections: Arc::new(RwLock::new(Vec::new())),
+        };
+
+        rho_connection
+    }
+
+    pub async fn get_iota_id(&self) -> u64 {
+        self.iota_connection.iota_id
+    }
+
+    pub async fn get_user_ids(&self) -> Vec<i64> {
+        self.user_ids.read().await.clone()
+    }
+
+    pub async fn set_user_ids(&self, user_ids: Vec<i64>) {
+        let mut guard = self.user_ids.write().await;
+        *guard = user_ids;
+    }
+
+    pub async fn add_user_id(&self, user_id: i64) {
+        let mut guard = self.user_ids.write().await;
+        if !guard.contains(&user_id) {
+            guard.push(user_id);
+        }
+    }
+
+    pub async fn bind_user_id(&self, user_id: i64) {
+        self.add_user_id(user_id).await;
+        self.iota_connection.add_user_id(user_id as u64).await;
+    }
+
+    pub fn get_iota_connection(&self) -> &Arc<IotaConnection> {
+        &self.iota_connection
+    }
+
+    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: i64,
+    ) -> 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 == user_id as u64 {
+                collections.push(con.clone());
+            }
+        }
+        collections
+    }
+
+    /// Add a client connection
+    #[allow(dead_code)]
+    pub async fn add_client_connection(&self, connection: Arc<ClientConnection>) {
+        let notification = CommunicationValue::new(CommunicationType::client_connected).add_data(
+            DataTypes::user_id,
+            DataValue::Number(connection.get_user_id().await as i64),
+        );
+
+        self.iota_connection.send_message(&notification).await;
+
+        {
+            let mut connections = self.client_connections.write().await;
+            connections.push(Arc::clone(&connection));
+        }
+
+        OmegaConnection::client_changed(
+            self.get_iota_id().await as i64,
+            connection.get_user_id().await as i64,
+            UserStatus::user_online,
+        )
+        .await;
+    }
+
+    /// Remove a client connection
+    pub async fn close_client_connection(&self, connection: Arc<ClientConnection>) {
+        let target_user_id = connection.get_user_id().await;
+        {
+            let mut connections = self.client_connections.write().await;
+            connections.retain(|con| {
+                futures::executor::block_on(async { con.get_user_id().await != target_user_id })
+            });
+        }
+
+        // Notify OmegaConnection
+        OmegaConnection::client_changed(
+            self.get_iota_id().await as i64,
+            connection.get_user_id().await as i64,
+            UserStatus::user_offline,
+        )
+        .await;
+    }
+
+    /// 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 as i64).await;
+
+        // Notify OmegaConnection
+        OmegaConnection::close_iota(self.get_iota_id().await as i64).await;
+    }
+
+    /// Send message from Iota to specific client
+    pub async fn message_to_client(&self, cv: CommunicationValue) {
+        let connections = self.client_connections.read().await;
+        let receiver_id = cv.get_receiver();
+        for connection in connections.iter() {
+            if connection.get_user_id().await == receiver_id {
+                connection.clone().send_message(&cv).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: i64, interested_ids: Vec<i64>) {
+        let connections = self.client_connections.read().await;
+        for connection in connections.iter() {
+            let conn_user_id = connection.get_user_id().await;
+            if conn_user_id == user_id as u64 {
+                connection
+                    .clone()
+                    .set_interested_users(interested_ids.clone())
+                    .await;
+                break;
+            }
+        }
+    }
+
+    /// Check if clients are interested in a user
+    #[allow(dead_code)]
+    pub async fn are_they_interested(&self, user_id: i64) {
+        let connections = self.client_connections.read().await;
+        for connection in connections.iter() {
+            connection.clone().are_you_interested(user_id).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() {
+            let 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
+    #[allow(dead_code)]
+    pub async fn contains_user(&self, user_id: &i64) -> bool {
+        self.user_ids.read().await.contains(user_id)
+    }
+
+    /// Get count of active client connections
+    #[allow(dead_code)]
+    pub async fn client_count(&self) -> usize {
+        let connections = self.client_connections.read().await;
+        connections.len()
+    }
+}

src/rho/rho_manager.rs

@@ -0,0 +1,83 @@
+use super::rho_connection::RhoConnection;
+use crate::log_in;
+use crate::util::logger::PrintType;
+use std::{
+    collections::HashMap,
+    sync::{Arc, LazyLock},
+};
+use tokio::sync::RwLock;
+
+pub static RHO_CONNECTIONS: LazyLock<Arc<RwLock<HashMap<i64, Arc<RhoConnection>>>>> =
+    LazyLock::new(|| Arc::new(RwLock::new(HashMap::new())));
+
+pub async fn get_rho_con_for_user(user_id: i64) -> Option<Arc<RhoConnection>> {
+    let connections = RHO_CONNECTIONS.read().await;
+    for rho_connection in connections.values() {
+        let rho_user_ids = rho_connection.get_user_ids().await;
+        log_in!(
+            user_id,
+            PrintType::Client,
+            "Comparing user IDs: {:?}",
+            rho_user_ids
+        );
+        if rho_user_ids.contains(&user_id) {
+            return Some(Arc::clone(rho_connection));
+        }
+    }
+    None
+}
+
+#[allow(dead_code)]
+pub async fn contains_iota(iota_id: i64) -> bool {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.contains_key(&iota_id)
+}
+
+/// Bind a user ID to an already tracked iota/rho connection.
+pub async fn bind_user_to_iota(user_id: i64, iota_id: i64) -> Option<Arc<RhoConnection>> {
+    let connections = RHO_CONNECTIONS.read().await;
+    if let Some(rho_connection) = connections.get(&iota_id) {
+        let rho = Arc::clone(rho_connection);
+        drop(connections);
+
+        rho.add_user_id(user_id).await;
+
+        log_in!(
+            user_id,
+            PrintType::Client,
+            "Bound user {} to iota {}",
+            user_id,
+            iota_id
+        );
+
+        Some(rho)
+    } else {
+        None
+    }
+}
+
+/// Remove a RhoConnection by Iota ID
+pub async fn remove_rho(iota_id: i64) -> 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 as i64, rho_connection);
+}
+
+/// Get a RhoConnection by Iota ID directly
+#[allow(dead_code)]
+pub async fn get_rho_by_iota(iota_id: i64) -> Option<Arc<RhoConnection>> {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.get(&iota_id).map(Arc::clone)
+}
+
+/// Get the count of active connections
+pub async fn connection_count() -> usize {
+    let connections = RHO_CONNECTIONS.read().await;
+    connections.len()
+}

src/rho/server.rs

@@ -0,0 +1,25 @@
+use crate::{
+    log,
+    rho::connection::GeneralConnection,
+    util::{file_util::load_file_vec, logger::PrintType},
+};
+use ttp_native::Host;
+
+pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
+    let cert_pem = load_file_vec("certs", "cert.pem").expect("Error loading Pemfile");
+
+    let key_pem = load_file_vec("certs", "key.pem").expect("Error loading Keyfile");
+
+    let mut host: Host = ttp_native::host(port, cert_pem, key_pem).await?;
+    log!(0, PrintType::General, "Server listening on port {}", port);
+
+    while let Some((sender, receiver)) = host.next().await {
+        tokio::spawn(async move {
+            let conn = GeneralConnection::new(sender, receiver);
+            conn.handle().await;
+        });
+    }
+    log!(0, PrintType::General, "Server stopped");
+
+    Ok(())
+}