From 7209e7ff578ce18f604adae06f18b431a4f61b35 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Sun, 12 Apr 2026 18:56:39 +0200 Subject: [PATCH] [Add] basic App connection --- Cargo.lock | 62 +++---- src/rho/app_connection.rs | 341 ++++++++++++++++++++++++++++++++++++++ src/rho/connection.rs | 5 + src/rho/mod.rs | 1 + src/rho/rho_connection.rs | 46 ++++- src/util/logger.rs | 2 + 6 files changed, 425 insertions(+), 32 deletions(-) create mode 100644 src/rho/app_connection.rs diff --git a/Cargo.lock b/Cargo.lock index cf9ebd3..ba102a2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -395,9 +395,9 @@ checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" [[package]] name = "cc" -version = "1.2.59" +version = "1.2.60" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7a4d3ec6524d28a329fc53654bbadc9bdd7b0431f5d65f1a56ffb28a1ee5283" +checksum = "43c5703da9466b66a946814e1adf53ea2c90f10063b86290cc9eb67ce3478a20" dependencies = [ "find-msvc-tools", "jobserver", @@ -1052,9 +1052,9 @@ dependencies = [ [[package]] name = "hashbrown" -version = "0.16.1" +version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" [[package]] name = "heck" @@ -1324,12 +1324,12 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.13.1" +version = "2.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "45a8a2b9cb3e0b0c1803dbb0758ffac5de2f425b23c28f518faabd9d805342ff" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", - "hashbrown 0.16.1", + "hashbrown 0.17.0", "serde", "serde_core", ] @@ -1395,9 +1395,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.94" +version = "0.3.95" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e04e2ef80ce82e13552136fabeef8a5ed1f985a96805761cbb9a2c34e7664d9" +checksum = "2964e92d1d9dc3364cae4d718d93f227e3abb088e747d92e0395bfdedf1c12ca" dependencies = [ "cfg-if", "futures-util", @@ -1490,7 +1490,7 @@ dependencies = [ "parking_lot", "pbjson-types", "prost", - "rand 0.9.2", + "rand 0.9.3", "reqwest", "scopeguard", "serde", @@ -2099,9 +2099,9 @@ dependencies = [ [[package]] name = "pkg-config" -version = "0.3.32" +version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" [[package]] name = "polling" @@ -2264,7 +2264,7 @@ dependencies = [ "bytes", "getrandom 0.3.4", "lru-slab", - "rand 0.9.2", + "rand 0.9.3", "ring", "rustc-hash", "rustls", @@ -2324,9 +2324,9 @@ dependencies = [ [[package]] name = "rand" -version = "0.9.2" +version = "0.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" +checksum = "7ec095654a25171c2124e9e3393a930bddbffdc939556c914957a4c3e0a87166" dependencies = [ "rand_chacha 0.9.0", "rand_core 0.9.5", @@ -2584,9 +2584,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.10" +version = "0.103.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef" +checksum = "20a6af516fea4b20eccceaf166e8aa666ac996208e8a644ce3ef5aa783bc7cd4" dependencies = [ "aws-lc-rs", "ring", @@ -3133,7 +3133,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "ttp-core" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#2322bdced8183e970405d5017cd87ba805309a4c" +source = "git+https://git.methanium.net/Tensamin/TTP.git#ff16a5a502ef9886104e1d1d08c22b0ad78e31ce" dependencies = [ "base64 0.22.1", "byteorder", @@ -3145,7 +3145,7 @@ dependencies = [ [[package]] name = "ttp-native" version = "0.1.0" -source = "git+https://git.methanium.net/Tensamin/TTP.git#2322bdced8183e970405d5017cd87ba805309a4c" +source = "git+https://git.methanium.net/Tensamin/TTP.git#ff16a5a502ef9886104e1d1d08c22b0ad78e31ce" dependencies = [ "quinn", "rustls", @@ -3324,9 +3324,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.117" +version = "0.2.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0551fc1bb415591e3372d0bc4780db7e587d84e2a7e79da121051c5c4b89d0b0" +checksum = "0bf938a0bacb0469e83c1e148908bd7d5a6010354cf4fb73279b7447422e3a89" dependencies = [ "cfg-if", "once_cell", @@ -3337,9 +3337,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.67" +version = "0.4.68" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "03623de6905b7206edd0a75f69f747f134b7f0a2323392d664448bf2d3c5d87e" +checksum = "f371d383f2fb139252e0bfac3b81b265689bf45b6874af544ffa4c975ac1ebf8" dependencies = [ "js-sys", "wasm-bindgen", @@ -3347,9 +3347,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.117" +version = "0.2.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7fbdf9a35adf44786aecd5ff89b4563a90325f9da0923236f6104e603c7e86be" +checksum = "eeff24f84126c0ec2db7a449f0c2ec963c6a49efe0698c4242929da037ca28ed" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3357,9 +3357,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.117" +version = "0.2.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dca9693ef2bab6d4e6707234500350d8dad079eb508dca05530c85dc3a529ff2" +checksum = "9d08065faf983b2b80a79fd87d8254c409281cf7de75fc4b773019824196c904" dependencies = [ "bumpalo", "proc-macro2", @@ -3370,9 +3370,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.117" +version = "0.2.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39129a682a6d2d841b6c429d0c51e5cb0ed1a03829d8b3d1e69a011e62cb3d3b" +checksum = "5fd04d9e306f1907bd13c6361b5c6bfc7b3b3c095ed3f8a9246390f8dbdee129" dependencies = [ "unicode-ident", ] @@ -3413,9 +3413,9 @@ dependencies = [ [[package]] name = "web-sys" -version = "0.3.94" +version = "0.3.95" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cd70027e39b12f0849461e08ffc50b9cd7688d942c1c8e3c7b22273236b4dd0a" +checksum = "4f2dfbb17949fa2088e5d39408c48368947b86f7834484e87b73de55bc14d97d" dependencies = [ "js-sys", "wasm-bindgen", diff --git a/src/rho/app_connection.rs b/src/rho/app_connection.rs new file mode 100644 index 0000000..0f59cbe --- /dev/null +++ b/src/rho/app_connection.rs @@ -0,0 +1,341 @@ +use crate::anonymous_clients::anonymous_manager; +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::{log_cv_in, log_cv_out, log_err, log_in, log_out}; +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 AppConnection { + pub user_id: u64, + pub app_identifier: String, + pub app_session: Uuid, + + pub sender: Arc, + pub receiver: Arc, + pub ping: Arc>, + pub_key: Arc>>>, + pub rho_connection: Arc>>>, + is_open: Arc>, +} + +impl AppConnection { + pub async fn from_general(general: Arc, user_id: u64) -> Arc { + Arc::new(Self { + ping: Arc::new(RwLock::new(0)), + pub_key: Arc::new(RwLock::new(None)), + rho_connection: general.rho_connection.clone(), + is_open: Arc::new(RwLock::new(true)), + sender: general.sender.clone(), + receiver: general.receiver.clone(), + user_id: user_id, + app_identifier: general.app_identifier.read().await.clone().unwrap(), + app_session: general.app_session.read().await.clone().unwrap(), + }) + } + pub fn start(self: Arc) { + 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> { + self.rho_connection.read().await.clone() + } + + /// Send a CommunicationValue to the app + pub async fn send_message(self: Arc, cv: &CommunicationValue) { + if !*self.is_open.read().await { + log_out!( + self.user_id as i64, + PrintType::App, + "Attempted to send message to a closed connection." + ); + return; + } + if !cv.is_type(CommunicationType::pong) && !cv.is_type(CommunicationType::ping) { + log_cv_out!(PrintType::App, &cv); + } + let _ = self.sender.send(&cv).await; + } + + /// Handle incoming message from app + pub async fn handle_message(self: Arc, cv: CommunicationValue) { + tokio::spawn(async move { + if cv.is_type(CommunicationType::ping) { + self.handle_ping(cv).await; + return; + } + log_cv_in!(PrintType::App, cv); + + 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::get_user_data) + || cv.is_type(CommunicationType::get_iota_data) + { + 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, cv: CommunicationValue) { + let app_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 { + app_for_closure.send_message(&response_cv).await; + } + }); + } + + /// Handle ping message + async fn handle_ping(self: Arc, 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; + } + + /// Forward message to Iota + async fn forward_to_iota(self: Arc, 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::App, + "Forwarding app->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::App, + "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::App, + "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::App, + "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::App, + "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, 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(); + } + + /// 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_app_connection(Arc::new(self.clone())).await; + } + } +} + +// Implement Clone to make it easier to work with Arc +impl Clone for AppConnection { + fn clone(&self) -> Self { + Self { + sender: Arc::clone(&self.sender), + receiver: Arc::clone(&self.receiver), + user_id: self.user_id, + app_identifier: self.app_identifier.clone(), + app_session: self.app_session, + ping: Arc::clone(&self.ping), + pub_key: Arc::clone(&self.pub_key), + rho_connection: Arc::clone(&self.rho_connection), + is_open: Arc::clone(&self.is_open), + } + } +} diff --git a/src/rho/connection.rs b/src/rho/connection.rs index cde8ee4..ff03450 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -3,6 +3,7 @@ use std::{sync::Arc, time::Duration}; use tokio::sync::RwLock; use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue, util::rand_u64}; use ttp_native::{Receiver, Sender}; +use uuid::Uuid; use crate::{ anonymous_clients::anonymous_client_connection::AnonymousClientConnection, @@ -41,6 +42,8 @@ pub struct GeneralConnection { pub rho_connection: Arc>>>, id: Arc>, pub session_id: Arc>, + pub app_identifier: Arc>>, + pub app_session: Arc>>, pub_key: Arc>>>, } @@ -57,6 +60,8 @@ impl GeneralConnection { rho_connection: Arc::new(RwLock::new(None)), id: Arc::new(RwLock::new(0)), session_id: Arc::new(RwLock::new(0)), + app_identifier: Arc::new(RwLock::new(None)), + app_session: Arc::new(RwLock::new(None)), pub_key: Arc::new(RwLock::new(None)), }) } diff --git a/src/rho/mod.rs b/src/rho/mod.rs index ac498f3..af516b1 100644 --- a/src/rho/mod.rs +++ b/src/rho/mod.rs @@ -1,3 +1,4 @@ +pub mod app_connection; pub mod client_connection; pub mod connection; pub mod iota_connection; diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 2a30430..534cf17 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,16 +1,18 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager}; -use crate::data::user::UserStatus; use crate::omega::omega_connection::OmegaConnection; +use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue}; +use uuid::Uuid; pub struct RhoConnection { iota_connection: Arc, user_ids: Arc>>, client_connections: Arc>>>, + app_connections: Arc>>>, } impl RhoConnection { @@ -20,6 +22,7 @@ impl RhoConnection { iota_connection, user_ids: Arc::new(RwLock::new(user_ids.clone())), client_connections: Arc::new(RwLock::new(Vec::new())), + app_connections: Arc::new(RwLock::new(Vec::new())), }; rho_connection @@ -74,6 +77,47 @@ impl RhoConnection { collections } + pub async fn get_app_connections( + &self, + userid: Option, + app_identifier: Option, + app_session: Option, + ) -> Vec> { + let connections = self.app_connections.read().await; + connections + .iter() + .filter(|conn| { + if let Some(uid) = userid { + if conn.user_id != uid as u64 { + return false; + } + } + if let Some(ref identifier) = app_identifier { + if conn.app_identifier != *identifier { + return false; + } + } + if let Some(ref session) = app_session { + if conn.app_session != *session { + return false; + } + } + true + }) + .cloned() + .collect() + } + + pub async fn add_app_connection(&self, connection: Arc) { + let mut connections = self.app_connections.write().await; + connections.push(connection); + } + + pub async fn close_app_connection(&self, connection: Arc) { + let mut connections = self.app_connections.write().await; + connections.retain(|c| c.app_session != connection.app_session); + } + /// Add a client connection #[allow(dead_code)] pub async fn add_client_connection(&self, connection: Arc) { diff --git a/src/util/logger.rs b/src/util/logger.rs index 9e9a085..42188f4 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -18,6 +18,7 @@ static LOGGER: OnceLock> = OnceLock::new(); pub enum PrintType { Call, Client, + App, Iota, Omikron, Omega, @@ -77,6 +78,7 @@ fn colorize(kind: PrintType, is_error: bool) -> Color { match kind { PrintType::Call => Color::Purple, PrintType::Client => Color::Green, + PrintType::App => Color::Green, PrintType::Iota => Color::Yellow, PrintType::Omikron => Color::Blue, PrintType::Omega => Color::Cyan,