From 87476579992d701ebc10c811190abb764bc16013 Mon Sep 17 00:00:00 2001 From: Alex-Emmet Date: Tue, 20 Jan 2026 21:45:05 +0100 Subject: [PATCH] anonymous calls --- .../anonymous_client_connection.rs | 547 ++++++++++++++++++ src/anonymous_clients/anonymous_manager.rs | 21 + src/anonymous_clients/mod.rs | 2 + src/calls/call_group.rs | 62 +- src/calls/call_manager.rs | 3 + src/calls/caller.rs | 5 +- src/data/communication.rs | 8 +- src/main.rs | 37 ++ src/rho/client_connection.rs | 28 +- src/rho/iota_connection.rs | 25 +- src/rho/rho_connection.rs | 12 +- src/util/crypto_util.rs | 1 - 12 files changed, 712 insertions(+), 39 deletions(-) create mode 100644 src/anonymous_clients/anonymous_client_connection.rs create mode 100644 src/anonymous_clients/anonymous_manager.rs create mode 100644 src/anonymous_clients/mod.rs diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs new file mode 100644 index 0000000..1ac7985 --- /dev/null +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -0,0 +1,547 @@ +use async_tungstenite::tungstenite::Message; +use async_tungstenite::{WebSocketReceiver, WebSocketSender}; +use json::JsonValue; +use json::number::Number; +use std::str::FromStr; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::RwLock; +use tokio_util::compat::Compat; +use tungstenite::Utf8Bytes; +use uuid::Uuid; + +use crate::calls::call_manager; +use crate::data::{ + communication::{CommunicationType, CommunicationValue, DataTypes}, + user::User, +}; +use crate::omega::omega_connection::{WAITING_TASKS, get_omega_connection}; +use crate::rho::rho_manager; +use crate::util::logger::PrintType; +use crate::{log_in, log_out}; + +pub struct AnonymousClientConnection { + pub sender: Arc>>>, + pub receiver: Arc>>>, + pub user_id: Arc>, + pub ping: Arc>, + pub interested_users: Arc>>, + is_open: Arc>, + pub user_name: Arc>, + pub display_name: Arc>, + pub avatar: Arc>, +} + +impl AnonymousClientConnection { + /// Create a new AnonymousClientConnection + pub fn new( + sender: WebSocketSender>, + receiver: WebSocketReceiver>, + ) -> Arc { + Arc::new(Self { + sender: Arc::new(RwLock::new(sender)), + receiver: Arc::new(RwLock::new(receiver)), + user_id: Arc::new(RwLock::new( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64, + )), + ping: Arc::new(RwLock::new(-1)), + interested_users: Arc::new(RwLock::new(Vec::new())), + is_open: Arc::new(RwLock::new(true)), + user_name: Arc::new(RwLock::new(String::new())), + display_name: Arc::new(RwLock::new(String::new())), + avatar: Arc::new(RwLock::new(String::new())), + }) + } + + /// Get the user ID + pub async fn get_user_id(&self) -> i64 { + *self.user_id.read().await + } + + /// Get the user name + pub async fn get_user_name(&self) -> String { + self.user_name.read().await.clone() + } + + /// Get the display name + pub async fn get_display_name(&self) -> String { + self.display_name.read().await.clone() + } + + /// Get the avatar + pub async fn get_avatar(&self) -> String { + self.avatar.read().await.clone() + } + + /// Send a string message to the client + pub async fn send_message_str(self: Arc, message: &str) { + let mut session = self.sender.write().await; + if let Err(e) = session + .send(Message::Text(Utf8Bytes::from(message.to_string()))) + .await + { + log_out!( + PrintType::Client, + "Failed to send message to anonymous client: {}", + e, + ); + } + } + + /// Send a CommunicationValue to the client + pub async fn send_message(self: Arc, cv: &CommunicationValue) { + if !*self.is_open.read().await { + log_out!( + PrintType::Client, + "Attempted to send message to a closed connection." + ); + return; + } + if !cv.is_type(CommunicationType::pong) { + log_out!(PrintType::Client, "{}", &cv.to_json().to_string()); + } + self.send_message_str(&cv.to_json().to_string()).await; + } + + /// Handle incoming message from client + pub async fn handle_message(self: Arc, message: Utf8Bytes) { + tokio::spawn(async move { + let cv = CommunicationValue::from_json(&message); + if cv.is_type(CommunicationType::ping) { + self.handle_ping(cv).await; + return; + } + log_in!( + PrintType::Client, + "Anonymous: {}", + &cv.to_json().to_string() + ); + + if cv.is_type(CommunicationType::identification) { + let call_id = Uuid::parse_str( + cv.get_data(DataTypes::call_id) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""), + ) + .unwrap_or(Uuid::new_v4()); + + let call = if let Some(call) = call_manager::get_call(call_id).await { + if call.is_anonymous().await { + call + } else { + self.send_error_response( + &cv.get_id(), + CommunicationType::error_not_authenticated, + ) + .await; + return; + } + } else { + self.send_error_response( + &cv.get_id(), + CommunicationType::error_not_authenticated, + ) + .await; + return; + }; + + let mut invited = JsonValue::new_array(); + for call_invitee in call.members.read().await.clone() { + let call_invitee_cv = get_omega_connection() + .await_response( + &CommunicationValue::new(CommunicationType::get_user_data).add_data( + DataTypes::user_id, + JsonValue::from(call_invitee.user_id), + ), + Some(Duration::from_secs(2)), + ) + .await + .unwrap(); + let mut json_invitee = JsonValue::new_object(); + let _ = json_invitee.insert( + "user_id", + call_invitee_cv + .get_data(DataTypes::user_id) + .unwrap_or(&JsonValue::Null) + .clone(), + ); + let _ = json_invitee.insert( + "username", + call_invitee_cv + .get_data(DataTypes::username) + .unwrap_or(&JsonValue::Null) + .clone(), + ); + let _ = json_invitee.insert( + "display", + call_invitee_cv + .get_data(DataTypes::display) + .unwrap_or(&JsonValue::Null) + .clone(), + ); + let _ = json_invitee.insert( + "avatar", + call_invitee_cv + .get_data(DataTypes::avatar) + .unwrap_or(&JsonValue::Null) + .clone(), + ); + + let _ = invited.push(json_invitee); + } + + let token = call.create_anonymous_token(self.get_user_id().await).await; + + let mut serialized = JsonValue::new_object(); + let _ = serialized.insert("call_id", JsonValue::String(call_id.to_string())); + let _ = serialized.insert("call_invited", invited.clone()); + let _ = serialized.insert("call_members", invited); + let _ = serialized.insert("call_token", JsonValue::String(token.unwrap())); + self.clone() + .send_message( + &&CommunicationValue::new(CommunicationType::identification_response) + .with_id(cv.get_id()) + .add_data( + DataTypes::user_id, + JsonValue::from(self.get_user_id().await), + ) + .add_data( + DataTypes::username, + JsonValue::String(self.clone().get_user_name().await), + ) + .add_data( + DataTypes::display, + JsonValue::String(self.get_display_name().await), + ) + .add_data( + DataTypes::avatar, + JsonValue::String(self.get_avatar().await), + ) + .add_data(DataTypes::call_state, serialized), + ) + .await; + } + + // 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::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::change_user_data) { + // TODO + return; + } + + if cv.is_type(CommunicationType::get_user_data) + || cv.is_type(CommunicationType::get_iota_data) + || cv.is_type(CommunicationType::delete_user) + { + self.handle_omega_forward(cv).await; + return; + } + }); + } + async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { + let client_for_closure = self.clone(); + WAITING_TASKS.insert( + cv.get_id(), + Box::new(move |_, response_cv| { + let client = client_for_closure.clone(); + tokio::spawn(async move { + client.send_message(&response_cv).await; + }); + true + }), + ); + get_omega_connection() + .send_message(&cv.with_sender(*self.user_id.read().await)) + .await; + } + + /// Handle ping message + async fn handle_ping(self: Arc, 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::() { + let mut ping_guard = self.ping.write().await; + *ping_guard = ping_val; + } + } + + // Send pong response + let response = CommunicationValue::new(CommunicationType::pong).with_id(cv.get_id()); + + self.send_message(&response).await; + } + + /// Handle client status change + async fn handle_client_changed(self: Arc, _cv: CommunicationValue) { + /*let user_id = self.get_user_id().await; + if let Some(_status_str) = cv.get_data(DataTypes::user_state) { + let user_status = UserStatus::online; + }*/ + } + + /// Handle call invite + async fn handle_call_invite(self: Arc, cv: CommunicationValue) { + let receiver_id: i64 = cv + .get_data(DataTypes::receiver_id) + .unwrap_or(&json::JsonValue::Number(Number::from(0))) + .as_i64() + .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) { + Some(id_str) => match Uuid::parse_str(&id_str.to_string()) { + Ok(id) => id, + Err(_) => { + self.send_error_response( + &cv.get_id(), + CommunicationType::error_invalid_call_id, + ) + .await; + return; + } + }, + None => { + 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.read().await, receiver_id).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, + None => { + 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) + .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()); + + 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, cv: CommunicationValue) { + let user_id = self.get_user_id().await; + + 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; + } + }; + + 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) + .add_data_str(DataTypes::call_token, 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, cv: CommunicationValue) { + let call_id = Uuid::from_str( + cv.get_data(DataTypes::call_id) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""), + ) + .unwrap(); + let user_id = cv + .get_data(DataTypes::user_id) + .unwrap_or(&JsonValue::Null) + .as_i64() + .unwrap_or(0); + let untill = cv + .get_data(DataTypes::untill) + .unwrap_or(&JsonValue::Null) + .as_i64() + .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.get_caller(user_id) + .await + .unwrap() + .set_timeout(untill) + .await; + } + } + } + async fn handle_call_disconnect_user(self: Arc, cv: CommunicationValue) { + let call_id = Uuid::from_str( + cv.get_data(DataTypes::call_id) + .unwrap_or(&JsonValue::Null) + .as_str() + .unwrap_or(""), + ) + .unwrap(); + let user_id = cv + .get_data(DataTypes::user_id) + .unwrap_or(&JsonValue::Null) + .as_i64() + .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).await; + } + } + } + + /// Send error response + async fn send_error_response( + self: Arc, + 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 is_open_guard = self.is_open.write().await; + if *is_open_guard { + return; + } + *is_open_guard = false; + + let mut session = self.sender.write().await; + let _ = session.close(None).await; + } + + /// Set interested users list + pub async fn set_interested_users(self: Arc, interested_ids: Vec) { + let mut interested_guard = self.interested_users.write().await; + *interested_guard = interested_ids; + } + pub async fn get_interested_users(self: Arc) -> Vec { + let interested_guard = self.interested_users.read().await; + interested_guard.clone() + } + + /// Check if interested in a user and send notification + pub async fn are_you_interested(self: Arc, user: &User) { + let interested_guard = self.clone().get_interested_users().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(¬ification).await; + } + } + + /// Handle connection close + pub async fn handle_close(&self) { + + // TODO delete temp user + } +} + +// Implement Clone to make it easier to work with Arc +impl Clone for AnonymousClientConnection { + fn clone(&self) -> Self { + Self { + sender: Arc::clone(&self.sender), + receiver: Arc::clone(&self.receiver), + user_id: Arc::clone(&self.user_id), + ping: Arc::clone(&self.ping), + interested_users: Arc::clone(&self.interested_users), + is_open: Arc::clone(&self.is_open), + user_name: Arc::clone(&self.user_name), + display_name: Arc::clone(&self.display_name), + avatar: Arc::clone(&self.avatar), + } + } +} diff --git a/src/anonymous_clients/anonymous_manager.rs b/src/anonymous_clients/anonymous_manager.rs new file mode 100644 index 0000000..320fa34 --- /dev/null +++ b/src/anonymous_clients/anonymous_manager.rs @@ -0,0 +1,21 @@ +use std::sync::Arc; + +use dashmap::DashMap; +use once_cell::sync::Lazy; + +use crate::anonymous_clients::anonymous_client_connection::AnonymousClientConnection; + +static ANONYMOUS_USERS: Lazy>> = + Lazy::new(|| DashMap::new()); + +pub async fn add_anonymous_user(connection: Arc) { + ANONYMOUS_USERS.insert(connection.get_user_id().await, connection); +} + +pub async fn remove_anonymous_user(user_id: i64) { + ANONYMOUS_USERS.remove(&user_id); +} + +pub async fn get_anonymous_user(user_id: i64) -> Option> { + ANONYMOUS_USERS.get(&user_id).map(|c| c.clone()) +} diff --git a/src/anonymous_clients/mod.rs b/src/anonymous_clients/mod.rs new file mode 100644 index 0000000..3edd0b4 --- /dev/null +++ b/src/anonymous_clients/mod.rs @@ -0,0 +1,2 @@ +pub mod anonymous_client_connection; +pub mod anonymous_manager; diff --git a/src/calls/call_group.rs b/src/calls/call_group.rs index 0f10fbe..04208fe 100644 --- a/src/calls/call_group.rs +++ b/src/calls/call_group.rs @@ -1,15 +1,20 @@ -use std::sync::Arc; - +use json::JsonValue; +use std::{env, sync::Arc, time::Duration}; use tokio::sync::RwLock; use uuid::Uuid; -use crate::calls::caller::Caller; +use crate::{ + calls::{call_util, caller::Caller}, + data::communication::{CommunicationType, CommunicationValue, DataTypes}, + omega::omega_connection::get_omega_connection, +}; pub struct CallGroup { pub call_id: Uuid, pub members: RwLock>>, pub show: RwLock, pub anonymous_joining: RwLock, + pub short_link: RwLock>, } impl CallGroup { @@ -19,6 +24,7 @@ impl CallGroup { members: RwLock::new(vec![user]), show: RwLock::new(true), anonymous_joining: RwLock::new(false), + short_link: RwLock::new(None), } } @@ -31,8 +37,54 @@ impl CallGroup { .cloned() } + pub async fn is_anonymous(&self) -> bool { + *self.anonymous_joining.read().await + } + pub async fn set_anonymous_joining(&self, enable: bool) { *self.anonymous_joining.write().await = enable; + + if self.short_link.read().await.is_none() { + let long_link = format!( + "https://app.tensamin.net/call/anonymous?call_id={}&omikron_id={}", + self.call_id, + env::var("ID") + .unwrap_or("0".to_string()) + .parse::() + .unwrap_or(0), + ); + let response_cv = get_omega_connection() + .await_response( + &CommunicationValue::new(CommunicationType::shorten_link) + .add_data(DataTypes::link, JsonValue::from(long_link)), + Some(Duration::from_secs(20)), + ) + .await; + if let Ok(response) = response_cv { + *self.short_link.write().await = Some( + response + .get_data(DataTypes::link) + .unwrap() + .as_str() + .unwrap() + .to_string(), + ); + log::info!( + "Shortened link for call {} is {}", + self.call_id, + self.short_link.read().await.as_ref().unwrap() + ); + } + } + } + + pub async fn create_anonymous_token(&self, user_id: i64) -> Option { + if self.is_anonymous().await { + if let Ok(token) = call_util::create_token(user_id, self.call_id, false) { + return Some(token); + } + } + None } pub async fn remove_caller(&self, user_id: i64) { @@ -41,4 +93,8 @@ impl CallGroup { .await .retain(|caller| caller.user_id != user_id); } + + pub async fn get_short_link(self: Arc) -> Option { + self.short_link.read().await.clone() + } } diff --git a/src/calls/call_manager.rs b/src/calls/call_manager.rs index b407d50..6bca183 100644 --- a/src/calls/call_manager.rs +++ b/src/calls/call_manager.rs @@ -50,6 +50,9 @@ pub async fn get_call_token(user_id: i64, call_id: Uuid) -> Option { } return Some(member.create_token()); } + if cg.is_anonymous().await { + return cg.create_anonymous_token(user_id).await; + } return None; } diff --git a/src/calls/caller.rs b/src/calls/caller.rs index f812e3f..7a6a600 100644 --- a/src/calls/caller.rs +++ b/src/calls/caller.rs @@ -1,7 +1,4 @@ -use std::{ - sync::Arc, - time::{SystemTime, UNIX_EPOCH}, -}; +use std::time::{SystemTime, UNIX_EPOCH}; use tokio::sync::RwLock; use uuid::Uuid; diff --git a/src/data/communication.rs b/src/data/communication.rs index fd57fe1..7c1660d 100644 --- a/src/data/communication.rs +++ b/src/data/communication.rs @@ -13,6 +13,9 @@ pub enum DataTypes { accepted_ids, uuid, register_id, + + link, + settings, settings_name, chat_partner_id, @@ -42,7 +45,7 @@ pub enum DataTypes { call_id, call_token, untill, - enable, + enabled, start_date, end_date, receiver_id, @@ -124,6 +127,9 @@ pub enum CommunicationType { error_no_call_id, error_invalid_call_id, success, + + shorten_link, + settings_save, settings_load, settings_list, diff --git a/src/main.rs b/src/main.rs index d7fa6ae..75a889a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,4 @@ +mod anonymous_clients; mod calls; mod data; mod omega; @@ -7,6 +8,7 @@ mod util; use async_tungstenite::accept_hdr_async; use dotenv::dotenv; use futures::StreamExt; +use json::JsonValue; use once_cell::sync::Lazy; use std::{env, sync::Arc}; use tokio::net::TcpListener; @@ -14,7 +16,9 @@ use tokio_util::compat::TokioAsyncReadCompatExt; use tungstenite::handshake::server::{Request, Response}; use crate::{ + anonymous_clients::anonymous_client_connection::AnonymousClientConnection, calls::call_manager::garbage_collect_calls, + data::communication::{CommunicationType, CommunicationValue, DataTypes}, omega::omega_connection::OmegaConnection, rho::{client_connection::ClientConnection, iota_connection::IotaConnection}, util::{ @@ -100,6 +104,39 @@ async fn main() { } } } + } else if path == "/ws/anonymous_client/" { + log_in!(PrintType::Client, "New Anonymous Client connection"); + let client_conn: Arc = + Arc::from(AnonymousClientConnection::new(sender, receiver)); + loop { + let msg_result = { + let mut session_lock = client_conn.receiver.write().await; + session_lock.next().await + }; + + match msg_result { + Some(Ok(msg)) => { + if msg.is_text() { + let text = msg.into_text().unwrap(); + client_conn.clone().handle_message(text).await; + } else if msg.is_close() { + log_in!(PrintType::Client, "Anonymous Client disconnected"); + client_conn.handle_close().await; + return; + } + } + Some(Err(e)) => { + log_err!(PrintType::Client, "WebSocket error: {}", e); + client_conn.handle_close().await; + return; + } + None => { + log_in!(PrintType::Client, "Anonymous Client stream ended"); + client_conn.handle_close().await; + return; + } + } + } } else if path == "/ws/iota/" { log_in!(PrintType::Iota, "New Iota connection"); let iota_conn: Arc = diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 470bc1a..4cd817f 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -273,12 +273,6 @@ impl ClientConnection { return; } - // Handle ping - if cv.is_type(CommunicationType::ping) { - self.handle_ping(cv).await; - return; - } - log_in!(PrintType::Client, "{}", &cv.to_json().to_string()); // Handle client status changes if cv.is_type(CommunicationType::client_changed) { self.handle_client_changed(cv).await; @@ -504,7 +498,11 @@ impl ClientConnection { .unwrap() .has_admin() { - call.get_caller(user_id).await.unwrap().set_timeout(untill); + call.get_caller(user_id) + .await + .unwrap() + .set_timeout(untill) + .await; } } } @@ -543,12 +541,14 @@ impl ClientConnection { ) .unwrap(); let enable = cv - .get_data(DataTypes::enable) + .get_data(DataTypes::enabled) .unwrap_or(&JsonValue::Null) .as_bool() - .unwrap_or(false); + .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) @@ -558,7 +558,17 @@ impl ClientConnection { { 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, JsonValue::String(call_id.to_string())) + .add_data(DataTypes::enabled, JsonValue::Boolean(enable)); + if let Some(short_link) = short_link { + response_cv = response_cv.add_data(DataTypes::link, JsonValue::String(short_link)); + } + self.send_message(&response_cv).await; } /// Forward message to Iota diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 06f0704..4583d3d 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -2,7 +2,6 @@ use crate::calls::call_group::CallGroup; use crate::calls::call_manager; use crate::get_private_key; use crate::get_public_key; -use crate::log; use crate::log_err; use crate::log_in; use crate::log_out; @@ -514,22 +513,22 @@ impl IotaConnection { for call in calls { for inviter in call.members.read().await.iter() { let call_self = call.get_caller(receiver_id).await.unwrap(); + let admin = call_self.has_admin(); let inviter_id = inviter.user_id; + let timeout = *call_self.timeout.read().await; + + let mut call_obj = JsonValue::new_object(); + let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string())); + if timeout > 0 { + let _ = call_obj.insert("timeout", JsonValue::from(timeout)); + } + if admin { + let _ = call_obj.insert("admin", JsonValue::Boolean(admin)); + } + if let Some(call_ids) = invites.get_mut(&inviter_id) { - let mut call_obj = JsonValue::new_object(); - let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string())); - let timeout = *call_self.timeout.read().await; - if timeout > 0 { - let _ = call_obj.insert("timeout", JsonValue::from(timeout)); - } call_ids.push(call_obj); } else { - let mut call_obj = JsonValue::new_object(); - let _ = call_obj.insert("call_id", JsonValue::String(call.call_id.to_string())); - let timeout = *call_self.timeout.read().await; - if timeout > 0 { - let _ = call_obj.insert("timeout", JsonValue::from(timeout)); - } invites.insert(inviter_id, vec![call_obj]); } } diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 50d6cfe..dfc71cc 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,13 +1,9 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection, rho_manager}; -use crate::omega::omega_connection::OmegaConnection; -use crate::util::logger::PrintType; -use crate::{ - data::{ - communication::{CommunicationType, CommunicationValue, DataTypes}, - user::UserStatus, - }, - log, +use crate::data::{ + communication::{CommunicationType, CommunicationValue, DataTypes}, + user::UserStatus, }; +use crate::omega::omega_connection::OmegaConnection; use json::{JsonValue, number::Number}; use std::collections::HashMap; use std::sync::Arc; diff --git a/src/util/crypto_util.rs b/src/util/crypto_util.rs index 066149a..97acab4 100644 --- a/src/util/crypto_util.rs +++ b/src/util/crypto_util.rs @@ -5,7 +5,6 @@ use aes_gcm::{ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STD}; use hkdf::Hkdf; use sha2::{Digest, Sha256}; -use std::fmt; use x448::{PublicKey, Secret}; // --- Custom Errors ---