tensamin/omikron
Watch
6
0
Fork
You've already forked omikron
 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

Swapped Id's to be Unix Timestamps instead of UUID's

Browse source
This commit is contained in:
Alex Emmet 2025-12-07 16:11:22 +01:00
commit b9208c190e
13 changed files with 352 additions and 291 deletions
Download patch file Download diff file Expand all files Collapse all files

115
src/rho/client_connection.rs
View file

@ -1,5 +1,6 @@
use async_tungstenite::tungstenite::Message;
use async_tungstenite::{WebSocketReceiver, WebSocketSender};
use json::number::Number;
use std::sync::{Arc, Weak};
use tokio::sync::RwLock;
use tokio_util::compat::Compat;
@ -27,7 +28,7 @@ pub struct ClientConnection {
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>,
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
/// User ID associated with this client
pub user_id: Arc<RwLock<Option<Uuid>>>,
pub user_id: Arc<RwLock<i64>>,
/// Whether this connection has been identified/authenticated
pub identified: Arc<RwLock<bool>>,
/// Ping latency tracking
@ -35,7 +36,7 @@ pub struct ClientConnection {
/// Weak reference to RhoConnection to avoid circular references
pub rho_connection: Arc<RwLock<Option<Weak<RhoConnection>>>>,
/// List of user IDs this client is interested in receiving updates about
pub interested_users: Arc<RwLock<Vec<Uuid>>>,
pub interested_users: Arc<RwLock<Vec<i64>>>,
}
impl ClientConnection {
@ -47,7 +48,7 @@ impl ClientConnection {
Arc::new(Self {
sender: Arc::new(RwLock::new(sender)),
receiver: Arc::new(RwLock::new(receiver)),
user_id: Arc::new(RwLock::new(None)),
user_id: Arc::new(RwLock::new(0)),
identified: Arc::new(RwLock::new(false)),
ping: Arc::new(RwLock::new(-1)),
rho_connection: Arc::new(RwLock::new(None)),
@ -56,7 +57,7 @@ impl ClientConnection {
}
/// Get the user ID
pub async fn get_user_id(&self) -> Option<Uuid> {
pub async fn get_user_id(&self) -> i64 {
*self.user_id.read().await
}
@ -155,18 +156,8 @@ impl ClientConnection {
/// Handle identification message
async fn handle_identification(&self, sarc: Arc<ClientConnection>, cv: CommunicationValue) {
// Extract user ID
let user_id = match cv.get_data(DataTypes::user_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id,
Err(_) => {
self.send_error_response(
&cv.get_id(),
CommunicationType::error_invalid_user_id,
)
.await;
return;
}
},
let user_id: i64 = match cv.get_data(DataTypes::user_id) {
Some(id_str) => id_str.as_i64().unwrap_or(0),
None => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_user_id)
.await;
@ -209,7 +200,7 @@ impl ClientConnection {
// Set identification data
{
let mut user_id_guard = self.user_id.write().await;
*user_id_guard = Some(user_id);
*user_id_guard = user_id;
}
{
let mut identified_guard = self.identified.write().await;
@ -253,61 +244,53 @@ impl ClientConnection {
/// Handle client status change
async fn handle_client_changed(&self, cv: CommunicationValue) {
if let Some(user_id) = self.get_user_id().await {
if let Some(_status_str) = cv.get_data(DataTypes::user_state) {
// Parse user status - this would need to be implemented properly
let user_status = UserStatus::online; // placeholder
if let Some(rho_conn) = self.get_rho_connection().await {
OmegaConnection::client_changed(
rho_conn.get_iota_id().await,
user_id,
user_status,
)
let user_id = self.get_user_id().await;
if let Some(_status_str) = cv.get_data(DataTypes::user_state) {
// Parse user status - this would need to be implemented properly
let user_status = UserStatus::online; // placeholder
if let Some(rho_conn) = self.get_rho_connection().await {
OmegaConnection::client_changed(rho_conn.get_iota_id().await, user_id, user_status)
.await;
}
}
}
}
/// Handle call invite
async fn handle_call_invite(&self, cv: CommunicationValue) {
let receiver_id = match cv.get_data(DataTypes::receiver_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id,
Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error)
.await;
return;
}
},
None => {
self.send_error_response(&cv.get_id(), CommunicationType::error)
.await;
return;
}
};
let 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)
.await;
self.send_error_response(
&cv.get_id(),
CommunicationType::error_invalid_call_id,
)
.await;
return;
}
},
None => {
self.send_error_response(&cv.get_id(), CommunicationType::error)
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.unwrap(), receiver_id)
.await;
if invited {
self.send_error_response(&cv.get_id(), CommunicationType::error)
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;
}
@ -323,10 +306,7 @@ impl ClientConnection {
};
// Get sender user ID
let sender_id = match self.get_user_id().await {
Some(id) => id,
None => return,
};
let sender_id = self.get_user_id().await;
// Create and send call distribution message
let forward = CommunicationValue::new(CommunicationType::call_invite)
@ -344,10 +324,7 @@ impl ClientConnection {
/// Handle get call request
async fn handle_get_call(&self, cv: CommunicationValue) {
let user_id = match self.get_user_id().await {
Some(id) => id,
None => return,
};
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()) {
@ -380,11 +357,10 @@ impl ClientConnection {
/// Forward message to Iota
async fn forward_to_iota(&self, cv: CommunicationValue) {
if let Some(user_id) = self.get_user_id().await {
if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv.with_sender(user_id);
rho_conn.message_to_iota(updated_cv).await;
}
let user_id = self.get_user_id().await;
if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv.with_sender(user_id);
rho_conn.message_to_iota(updated_cv).await;
}
}
@ -401,7 +377,7 @@ impl ClientConnection {
}
/// Set interested users list
pub async fn set_interested_users(&self, interested_ids: Vec<Uuid>) {
pub async fn set_interested_users(&self, interested_ids: Vec<i64>) {
let mut interested_guard = self.interested_users.write().await;
*interested_guard = interested_ids;
}
@ -424,12 +400,11 @@ impl ClientConnection {
/// Handle connection close
pub async fn handle_close(&self) {
if self.is_identified().await {
if let Some(user_id) = self.get_user_id().await {
if let Some(rho_conn) = rho_manager::get_rho_con_for_user(user_id).await {
rho_conn
.close_client_connection(Arc::new(self.clone()))
.await;
}
let user_id = self.get_user_id().await;
if let Some(rho_conn) = rho_manager::get_rho_con_for_user(user_id).await {
rho_conn
.close_client_connection(Arc::new(self.clone()))
.await;
}
}
}

102
src/rho/iota_connection.rs
View file

@ -7,6 +7,7 @@ use async_tungstenite::WebSocketReceiver;
use async_tungstenite::WebSocketSender;
use async_tungstenite::tungstenite::Message;
use json::JsonValue;
use json::number::Number;
use std::{
collections::HashMap,
sync::{Arc, Weak},
@ -27,8 +28,8 @@ use crate::{
pub struct IotaConnection {
pub sender: Arc<RwLock<WebSocketSender<Compat<tokio::net::TcpStream>>>>,
pub receiver: Arc<RwLock<WebSocketReceiver<Compat<tokio::net::TcpStream>>>>,
pub iota_id: Arc<RwLock<Uuid>>,
pub user_ids: Arc<RwLock<Vec<Uuid>>>,
pub iota_id: Arc<RwLock<i64>>,
pub user_ids: Arc<RwLock<Vec<i64>>>,
pub identified: Arc<RwLock<bool>>,
pub ping: Arc<RwLock<i64>>,
pub rho_connection: Arc<RwLock<Option<Weak<RhoConnection>>>>,
@ -43,7 +44,7 @@ impl IotaConnection {
Arc::new(Self {
sender: Arc::new(RwLock::new(sender)),
receiver: Arc::new(RwLock::new(receiver)),
iota_id: Arc::new(RwLock::new(Uuid::nil())),
iota_id: Arc::new(RwLock::new(0)),
user_ids: Arc::new(RwLock::new(Vec::new())),
identified: Arc::new(RwLock::new(false)),
ping: Arc::new(RwLock::new(0)),
@ -52,12 +53,12 @@ impl IotaConnection {
}
/// Get the Iota ID
pub async fn get_iota_id(&self) -> Uuid {
pub async fn get_iota_id(&self) -> i64 {
*self.iota_id.read().await
}
/// Get the user IDs
pub async fn get_user_ids(&self) -> Vec<Uuid> {
pub async fn get_user_ids(&self) -> Vec<i64> {
self.user_ids.read().await.clone()
}
@ -115,6 +116,7 @@ impl IotaConnection {
// Handle identification
if cv.is_type(CommunicationType::identification) && !self.is_identified().await {
line(PrintType::IotaIn, &cv.to_json().to_string());
self.handle_identification(cv).await;
return;
}
@ -151,35 +153,47 @@ impl IotaConnection {
/// Handle identification message
async fn handle_identification(self: Arc<Self>, cv: CommunicationValue) {
// Parse Iota ID
let iota_id = match cv.get_data(DataTypes::iota_id) {
Some(id_str) => match Uuid::parse_str(&id_str.to_string()) {
Ok(id) => id,
Err(_) => {
self.send_error_response(&cv.get_id()).await;
return;
}
},
None => {
self.send_error_response(&cv.get_id()).await;
return;
}
};
let iota_id: i64 = cv
.get_data(DataTypes::iota_id)
.unwrap_or(&JsonValue::Number(Number::from(0)))
.as_i64()
.unwrap_or(0);
if iota_id == 0 {
let error = CommunicationValue::new(CommunicationType::error).with_id(cv.get_id());
self.send_message(error).await;
return;
}
// Parse user IDs
let mut validated_user_ids: Vec<Uuid> = Vec::new();
let mut validated_user_ids: Vec<i64> = Vec::new();
if let Some(user_ids_str) = cv.get_data(DataTypes::user_ids) {
for id_str in user_ids_str.to_string().split(',') {
if id_str.is_empty() {
continue;
}
if let Ok(user_id) = Uuid::parse_str(id_str.trim()) {
if let Some(auth_iota_id) = auth_connector::get_iota_id(user_id).await {
if auth_iota_id == iota_id {
validated_user_ids.push(user_id);
match id_str.parse::<i64>() {
Ok(user_id) => {
if let Some(auth_iota_id) = auth_connector::get_iota_id(user_id).await {
line(
PrintType::IotaIn,
&format!(
"auth for {} should be {} is {}",
user_id, iota_id, auth_iota_id
),
);
if auth_iota_id == iota_id {
validated_user_ids.push(user_id);
}
} else {
line(
PrintType::IotaIn,
&format!("User ID {} not parsed", id_str.trim()),
);
}
} else {
line(PrintType::IotaIn, "User ID not found");
}
Err(e) => {
line(
PrintType::IotaIn,
&format!("Failed to parse '{}' as i64: {:?}", id_str, e),
);
}
}
}
@ -280,10 +294,10 @@ impl IotaConnection {
/// Handle GET_CHATS message
async fn handle_get_chats(&self, cv: CommunicationValue) {
let receiver_id = cv.get_receiver();
let mut interested_ids: Vec<Uuid> = Vec::new();
let mut interested_ids: Vec<i64> = Vec::new();
let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(receiver_id).await;
let mut invites: HashMap<Uuid, Vec<JsonValue>> = HashMap::new();
let mut invites: HashMap<i64, Vec<JsonValue>> = HashMap::new();
let empty = &calls.is_empty();
for call in calls {
for inviter in call.members.read().await.iter() {
@ -309,22 +323,19 @@ impl IotaConnection {
JsonValue::new_array()
}
} else {
line(PrintType::CallIn, &format!("not empty: {:?}", invites));
let mut enrc_contacts = JsonValue::new_array();
if let Some(contacts_data) = cv.get_data(DataTypes::user_ids) {
if let JsonValue::Array(user_ids) = contacts_data {
for user_json in user_ids {
let user_id_str = user_json["user_id"].as_str().unwrap_or("");
if let Ok(user_id) = Uuid::parse_str(&user_id_str) {
interested_ids.push(user_id);
let mut enriched_contact = JsonValue::new_object();
let _ = enriched_contact.insert("user_id", user_id.to_string());
let _ = enriched_contact.insert(
"calls",
JsonValue::Array(invites.get(&user_id).unwrap_or(&vec![]).clone()),
);
let _ = enrc_contacts.push(enriched_contact);
}
let user_id = user_json["user_id"].as_i64().unwrap_or(0);
interested_ids.push(user_id);
let mut enriched_contact = JsonValue::new_object();
let _ = enriched_contact.insert("user_id", user_id.to_string());
let _ = enriched_contact.insert(
"calls",
JsonValue::Array(invites.get(&user_id).unwrap_or(&vec![]).clone()),
);
let _ = enrc_contacts.push(enriched_contact);
}
} else {
enrc_contacts = contacts_data.clone();
@ -355,11 +366,6 @@ impl IotaConnection {
}
}
async fn send_error_response(&self, message_id: &Uuid) {
let error = CommunicationValue::new(CommunicationType::error).with_id(*message_id);
self.send_message(error).await;
}
pub async fn handle_close(&self) {
if self.is_identified().await {
if let Some(rho_conn) = self.get_rho_connection().await {

71
src/rho/rho_connection.rs
View file

@ -4,20 +4,20 @@ use crate::data::{
user::UserStatus,
};
use crate::omega::omega_connection::OmegaConnection;
use json::{JsonValue, number::Number};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use uuid::Uuid;
pub struct RhoConnection {
iota_connection: Arc<IotaConnection>,
user_ids: Vec<Uuid>,
user_ids: 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<Uuid>) -> Self {
pub async fn new(iota_connection: Arc<IotaConnection>, user_ids: Vec<i64>) -> Self {
let rho_connection = Self {
iota_connection,
user_ids: user_ids.clone(),
@ -30,11 +30,11 @@ impl RhoConnection {
rho_connection
}
pub async fn get_iota_id(&self) -> Uuid {
pub async fn get_iota_id(&self) -> i64 {
self.iota_connection.get_iota_id().await
}
pub fn get_user_ids(&self) -> &Vec<Uuid> {
pub fn get_user_ids(&self) -> &Vec<i64> {
&self.user_ids
}
@ -50,12 +50,12 @@ impl RhoConnection {
/// Get client connections for a specific user
pub async fn get_client_connections_for_user(
&self,
user_id: Uuid,
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.unwrap() == user_id {
if con.get_user_id().await == user_id {
collections.push(con.clone());
}
}
@ -64,15 +64,10 @@ impl RhoConnection {
/// Add a client connection
pub async fn add_client_connection(&self, connection: Arc<ClientConnection>) {
let notification = CommunicationValue::new(CommunicationType::client_connected)
.add_data_str(
DataTypes::user_id,
connection
.get_user_id()
.await
.unwrap_or(Uuid::nil())
.to_string(),
);
let notification = CommunicationValue::new(CommunicationType::client_connected).add_data(
DataTypes::user_id,
JsonValue::Number(Number::from(connection.get_user_id().await)),
);
self.iota_connection.send_message(notification).await;
@ -83,7 +78,7 @@ impl RhoConnection {
OmegaConnection::client_changed(
self.get_iota_id().await,
connection.get_user_id().await.unwrap_or(Uuid::nil()),
connection.get_user_id().await,
UserStatus::online,
)
.await;
@ -94,12 +89,10 @@ impl RhoConnection {
{
let mut connections = self.client_connections.write().await;
let target_user_id = connection.get_user_id().await.unwrap();
let target_user_id = connection.get_user_id().await;
connections.retain(|con| {
futures::executor::block_on(async {
con.get_user_id().await.unwrap() != target_user_id
})
futures::executor::block_on(async { con.get_user_id().await != target_user_id })
});
connections.push(Arc::clone(&connection));
@ -108,7 +101,7 @@ impl RhoConnection {
// Notify OmegaConnection
OmegaConnection::client_changed(
self.get_iota_id().await,
connection.get_user_id().await.unwrap_or(Uuid::nil()),
connection.get_user_id().await,
UserStatus::user_offline,
)
.await;
@ -131,15 +124,9 @@ impl RhoConnection {
/// Send message from Iota to specific client
pub async fn message_to_client(&self, cv: CommunicationValue) {
if let Some(receiver_id) = Some(cv.get_receiver()) {
let connections = self.client_connections.read().await;
for connection in connections.iter() {
if let Some(conn_user_id) = connection.get_user_id().await {
if conn_user_id == receiver_id {
connection.send_message(&cv).await;
}
}
}
let connections = self.client_connections.read().await;
for connection in connections.iter() {
connection.send_message(&cv).await;
}
}
@ -154,16 +141,15 @@ impl RhoConnection {
}
/// Set interested users for a specific client
pub async fn set_interested(&self, user_id: Uuid, interested_ids: Vec<Uuid>) {
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() {
if let Some(conn_user_id) = connection.get_user_id().await {
if conn_user_id == user_id {
connection
.set_interested_users(interested_ids.clone())
.await;
break;
}
let conn_user_id = connection.get_user_id().await;
if conn_user_id == user_id {
connection
.set_interested_users(interested_ids.clone())
.await;
break;
}
}
}
@ -182,16 +168,15 @@ impl RhoConnection {
let mut pings = HashMap::new();
for connection in connections.iter() {
if let Some(user_id) = connection.get_user_id().await {
pings.insert(user_id.to_string(), connection.get_ping().await);
}
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
pub fn contains_user(&self, user_id: &Uuid) -> bool {
pub fn contains_user(&self, user_id: &i64) -> bool {
self.user_ids.contains(user_id)
}

11
src/rho/rho_manager.rs
View file

@ -6,12 +6,11 @@ use std::{
sync::{Arc, LazyLock},
};
use tokio::sync::RwLock;
use uuid::Uuid;
pub static RHO_CONNECTIONS: LazyLock<Arc<RwLock<HashMap<Uuid, Arc<RhoConnection>>>>> =
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: Uuid) -> Option<Arc<RhoConnection>> {
pub async fn get_rho_con_for_user(user_id: i64) -> Option<Arc<RhoConnection>> {
let connections = RHO_CONNECTIONS.read().await;
line(
PrintType::ClientIn,
@ -32,13 +31,13 @@ pub async fn get_rho_con_for_user(user_id: Uuid) -> Option<Arc<RhoConnection>> {
None
}
pub async fn contains_iota(iota_id: Uuid) -> bool {
pub async fn contains_iota(iota_id: i64) -> bool {
let connections = RHO_CONNECTIONS.read().await;
connections.contains_key(&iota_id)
}
/// Remove a RhoConnection by Iota ID
pub async fn remove_rho(iota_id: Uuid) -> Option<Arc<RhoConnection>> {
pub async fn remove_rho(iota_id: i64) -> Option<Arc<RhoConnection>> {
let mut connections = RHO_CONNECTIONS.write().await;
connections.remove(&iota_id)
}
@ -51,7 +50,7 @@ pub async fn add_rho(rho_connection: Arc<RhoConnection>) {
}
/// Get a RhoConnection by Iota ID directly
pub async fn get_rho_by_iota(iota_id: Uuid) -> Option<Arc<RhoConnection>> {
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)
}