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

[Fix] Calls, Spelling

Browse source
This commit is contained in:
Alex Emmet 2026-05-03 15:02:15 +02:00
commit fc85e9377e
10 changed files with 246 additions and 177 deletions
Download patch file Download diff file Expand all files Collapse all files

72
Cargo.lock generated
View file

@ -2,40 +2,6 @@
# It is not intended for manual editing.
version = 4
[[package]]
name = "Omikron"
version = "0.1.0"
dependencies = [
"aes-gcm",
"ansi_term",
"base64 0.22.1",
"dashmap",
"dotenv",
"futures",
"hex",
"hickory-resolver",
"hkdf 0.13.0",
"json",
"livekit-api",
"livekit-protocol",
"log",
"once_cell",
"rand 0.8.6",
"rand_core 0.6.4",
"rustls",
"serde",
"serde_json",
"sha2 0.11.0",
"strum",
"strum_macros",
"thiserror 2.0.18",
"tokio",
"ttp-core",
"ttp-native",
"uuid",
"x448",
]
[[package]]
name = "aead"
version = "0.5.2"
@ -2152,6 +2118,40 @@ dependencies = [
"asn1-rs",
]
[[package]]
name = "omikron"
version = "0.1.0"
dependencies = [
"aes-gcm",
"ansi_term",
"base64 0.22.1",
"dashmap",
"dotenv",
"futures",
"hex",
"hickory-resolver",
"hkdf 0.13.0",
"json",
"livekit-api",
"livekit-protocol",
"log",
"once_cell",
"rand 0.8.6",
"rand_core 0.6.4",
"rustls",
"serde",
"serde_json",
"sha2 0.11.0",
"strum",
"strum_macros",
"thiserror 2.0.18",
"tokio",
"ttp-core",
"ttp-native",
"uuid",
"x448",
]
[[package]]
name = "once_cell"
version = "1.21.4"
@ -3461,7 +3461,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
[[package]]
name = "ttp-core"
version = "0.1.0"
source = "git+https://git.methanium.net/Tensamin/TTP.git#9677eb6c6040e4e8d85ba8b6232aa954b5a9438d"
source = "git+https://git.methanium.net/Tensamin/TTP.git#0cc917d75562953b49a83804f9f37e89aca6fd88"
dependencies = [
"base64 0.22.1",
"byteorder",
@ -3474,7 +3474,7 @@ dependencies = [
[[package]]
name = "ttp-native"
version = "0.1.0"
source = "git+https://git.methanium.net/Tensamin/TTP.git#9677eb6c6040e4e8d85ba8b6232aa954b5a9438d"
source = "git+https://git.methanium.net/Tensamin/TTP.git#0cc917d75562953b49a83804f9f37e89aca6fd88"
dependencies = [
"quinn",
"rustls",

6
Cargo.toml
View file

@ -1,5 +1,5 @@
[package]
name = "Omikron"
name = "omikron"
version = "0.1.0"
edition = "2024"
@ -31,8 +31,8 @@ dotenv = "0.15.0"
hkdf = "0.13"
strum = "0.28.0"
strum_macros = "0.28.0"
livekit-api = { version = "0.4.14", features = ["native-tls"] }
livekit-protocol = "0.7.1"
livekit-api = { version = "0.4.19", features = ["native-tls"] }
livekit-protocol = "0.7.5"
thiserror = "2.0.18"
hickory-resolver = "0.25.2"
serde = "1.0.228"

8
src/anonymous_clients/anonymous_client_connection.rs
View file

@ -344,7 +344,12 @@ impl AnonymousClientConnection {
}
};
let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await;
let secret = cv
.get_data(DataTypes::call_secret)
.as_str()
.map(|s| s.to_string());
let invited =
call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
if !invited {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_call_id)
.await;
@ -459,6 +464,7 @@ impl AnonymousClientConnection {
self.send_message(&error).await;
}
/// Close the connection
#[allow(unused)]
pub async fn close(&self) {
let mut is_open_guard = self.is_open.write().await;
if !*is_open_guard {

37
src/calls/call_manager.rs
View file

@ -3,7 +3,7 @@ use once_cell::sync::Lazy;
use std::sync::Arc;
use uuid::Uuid;
use crate::calls::{call_group::CallGroup, caller::Caller};
use crate::calls::{call_group::CallGroup, call_util, caller::Caller};
pub static CALL_GROUPS: Lazy<DashMap<Uuid, Arc<CallGroup>>> = Lazy::new(|| DashMap::new());
#[allow(dead_code)]
@ -59,29 +59,23 @@ pub async fn get_call_token(user_id: u64, call_id: Uuid) -> Option<String> {
return Some(token);
}
if let Some(cg) = CALL_GROUPS.get(&call_id) {
let cg_clone = cg.clone();
let mut members = cg_clone.members.write().await;
if let Some(member) = members.iter().find(|m| m.user_id == user_id) {
return Some(member.create_token());
}
let new_caller = Arc::new(Caller::new(user_id, call_id, false));
let token = new_caller.create_token();
members.push(new_caller);
return Some(token);
}
let caller = Arc::new(Caller::new(user_id, call_id, true));
let call_group = CallGroup::new(call_id, caller.clone());
call_group.update_admins().await;
let call_group = Arc::new(CallGroup::new(call_id, caller.clone()));
CALL_GROUPS.insert(call_id, Arc::new(call_group));
CALL_GROUPS.insert(call_id, call_group.clone());
let _ = call_util::create_room(call_id).await;
call_group.update_admins().await;
Some(caller.create_token())
}
pub async fn add_invite(call_id: Uuid, inviter_id: u64, invitee_id: u64) -> bool {
pub async fn add_invite(
call_id: Uuid,
inviter_id: u64,
invitee_id: u64,
secret: Option<String>,
) -> bool {
if let Some(cg) = CALL_GROUPS.get(&call_id) {
let mut members = cg.members.write().await;
@ -91,6 +85,13 @@ pub async fn add_invite(call_id: Uuid, inviter_id: u64, invitee_id: u64) -> bool
if !members.iter().any(|m| m.user_id == invitee_id) {
members.push(Arc::new(Caller::new(invitee_id, call_id, false)));
}
if let Some(secret) = secret {
let mut secrets = cg.secrets.write().await;
secrets.insert((inviter_id, invitee_id), secret.clone());
secrets.insert((invitee_id, inviter_id), secret);
}
return true;
}
}

15
src/calls/call_util.rs
View file

@ -1,3 +1,4 @@
use livekit_api::services::room::CreateRoomOptions;
use livekit_api::{
access_token::{self},
services::room::RoomClient,
@ -35,6 +36,20 @@ pub fn get_livekit() -> Result<(String, String, String), ()> {
Ok((hostname, api_key, api_secret))
}
pub async fn create_room(call_id: Uuid) -> Result<(), ()> {
let (hostname, api_key, api_secret) = get_livekit()?;
let room_service = RoomClient::with_api_key(&hostname, &api_key, &api_secret);
let options = CreateRoomOptions::default();
room_service
.create_room(&call_id.to_string(), options)
.await
.map_err(|_| ())?;
Ok(())
}
pub fn create_token(user_id: u64, call_id: Uuid, has_admin: bool) -> Result<String, ()> {
let (_, api_key, api_secret) = get_livekit()?;

2
src/rho/app_connection.rs
View file

@ -64,6 +64,7 @@ impl AppConnection {
}
/// Get current ping
#[allow(unused)]
pub async fn get_ping(&self) -> i64 {
*self.ping.read().await
}
@ -304,6 +305,7 @@ impl AppConnection {
}
/// Close the connection
#[allow(unused)]
pub async fn close(&self) {
let mut is_open_guard = self.is_open.write().await;
if !*is_open_guard {

7
src/rho/client_connection.rs
View file

@ -274,7 +274,12 @@ impl ClientConnection {
}
};
let invited = call_manager::add_invite(call_id, self.user_id, receiver_id as u64).await;
let secret = cv
.get_data(DataTypes::call_secret)
.as_str()
.map(|s| s.to_string());
let invited =
call_manager::add_invite(call_id, self.user_id, receiver_id as u64, secret).await;
if !invited {
self.send_error_response(cv.get_id(), CommunicationType::error_invalid_call_id)
.await;

197
src/rho/connection.rs
View file

@ -1,5 +1,5 @@
use rand::{Rng, distributions::Alphanumeric};
use std::{sync::Arc, time::Duration};
use std::{collections::BTreeMap, collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::RwLock;
use ttp_core::{CommunicationType, CommunicationValue, DataTypes, DataValue};
use ttp_native::{Receiver, Sender};
@ -7,7 +7,7 @@ use uuid::Uuid;
use crate::{
anonymous_clients::anonymous_client_connection::AnonymousClientConnection,
calls::{call_group::CallGroup, call_manager, caller::Caller},
calls::call_manager,
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::{
@ -163,7 +163,7 @@ impl GeneralConnection {
.unwrap()
.export(DataFormat::Base64);
let response = CommunicationValue::new(CommunicationType::app_challange)
let response = CommunicationValue::new(CommunicationType::app_challenge)
.with_id(cv.get_id())
.add_data(
DataTypes::public_key,
@ -315,7 +315,7 @@ impl GeneralConnection {
.export(DataFormat::Base64);
let challenge_type = if cv.is_type(CommunicationType::app_identification) {
CommunicationType::app_challange
CommunicationType::app_challenge
} else {
CommunicationType::challenge
};
@ -337,7 +337,7 @@ impl GeneralConnection {
let id = *self.id.read().await as i64;
if !cv.is_type(CommunicationType::challenge_response)
&& !cv.is_type(CommunicationType::app_challange_response)
&& !cv.is_type(CommunicationType::app_challenge_response)
{
return;
}
@ -433,107 +433,108 @@ impl GeneralConnection {
if let Some(contacts) = v.as_array() {
let call_groups =
call_manager::get_call_groups(user_id as u64).await;
let callers =
call_manager::get_call_invites(user_id as u64).await;
let mapped: Vec<(Arc<Caller>, &Arc<CallGroup>)> = {
callers
.iter()
.map(|caller| {
(
Arc::clone(caller),
call_groups.iter().find(|call_group| {
call_group.call_id == caller.call_id
}),
)
})
.filter(|(_, call_group)| call_group.is_some())
.map(|(caller, call_group)| {
(caller, call_group.unwrap())
})
.collect()
};
if mapped.is_empty() {
DataValue::Array(contacts)
} else {
let mut new_contacts: Vec<DataValue> = Vec::new();
for contact in contacts {
if let Some(mut contact_map) = contact.as_map() {
let filtered_mapped: Vec<(
Arc<Caller>,
&Arc<CallGroup>,
)> = mapped
.clone()
.into_iter()
.filter(|(caller, _)| {
caller.user_id
== contact_map
.get(&DataTypes::user_id)
.unwrap_or(&DataValue::Null)
.as_number()
.unwrap_or(0)
as u64
})
.collect();
if filtered_mapped.is_empty() {
new_contacts.push(
DataValue::container_from_map(&contact_map),
);
let mut invites: HashMap<i64, Vec<DataValue>> = HashMap::new();
let mut global_calls: Vec<DataValue> = Vec::new();
for call in call_groups {
let members = call.members.read().await;
let call_self =
members.iter().find(|m| m.user_id == user_id as u64);
if let Some(call_self) = call_self {
let timeout = *call_self.timeout.read().await;
let admin = call_self.has_admin();
// List of all members in the call
let member_ids: Vec<DataValue> = members
.iter()
.map(|m| DataValue::Number(m.user_id as i64))
.collect();
let mut base_call_map: BTreeMap<DataTypes, DataValue> =
BTreeMap::new();
base_call_map.insert(
DataTypes::call_id,
DataValue::Str(call.call_id.to_string()),
);
base_call_map.insert(
DataTypes::call_members,
DataValue::Array(member_ids),
);
if timeout > 0 {
base_call_map.insert(
DataTypes::timeout,
DataValue::Number(timeout as i64),
);
}
if admin {
base_call_map.insert(
DataTypes::has_admin,
DataValue::Bool(true),
);
}
// Add to global calls (without contact-specific secret)
global_calls.push(DataValue::container_from_map(
&base_call_map,
));
for member in members.iter() {
let member_id = member.user_id;
if member_id == user_id as u64 {
continue;
}
let mut vec_of_filtered_calls: Vec<DataValue> =
Vec::new();
for (caller, call_group) in filtered_mapped {
vec_of_filtered_calls.push(
DataValue::Container(vec![
(
DataTypes::call_id,
DataValue::Str(
caller.call_id.to_string(),
),
),
(
DataTypes::call_secret,
DataValue::Str(
call_group
.secrets
.read()
.await
.get(&(
contact_map
.get(
&DataTypes::user_id,
)
.unwrap_or(
&DataValue::Null,
)
.as_number()
.unwrap_or(0)
as u64,
user_id as u64,
))
.unwrap_or(&"".to_string())
.clone(),
),
),
]),
let mut contact_call_map = base_call_map.clone();
if let Some(secret) = call
.secrets
.read()
.await
.get(&(member_id, user_id as u64))
{
contact_call_map.insert(
DataTypes::call_secret,
DataValue::Str(secret.clone()),
);
}
if !vec_of_filtered_calls.is_empty() {
contact_map.insert(
DataTypes::calls,
DataValue::Array(vec_of_filtered_calls),
);
}
new_contacts.push(DataValue::container_from_map(
&contact_map,
));
} else {
new_contacts.push(contact.clone());
invites
.entry(member_id as i64)
.or_insert_with(Vec::new)
.push(DataValue::container_from_map(
&contact_call_map,
));
}
}
DataValue::Array(new_contacts)
}
let mut new_contacts: Vec<DataValue> = Vec::new();
for contact in contacts {
if let Some(mut contact_map) = contact.as_map() {
if let Some(DataValue::Number(id)) =
contact_map.get(&DataTypes::user_id)
{
if let Some(call_list) = invites.get(id) {
contact_map.insert(
DataTypes::calls,
DataValue::Array(call_list.clone()),
);
}
}
new_contacts
.push(DataValue::container_from_map(&contact_map));
} else {
new_contacts.push(contact.clone());
}
}
ident_resp = ident_resp
.add_data(DataTypes::calls, DataValue::Array(global_calls));
DataValue::Array(new_contacts)
} else {
v.clone()
}

88
src/rho/iota_connection.rs
View file

@ -316,44 +316,82 @@ impl IotaConnection {
/// Handle GET_CHATS message
async fn handle_get_chats(&self, cv: CommunicationValue) {
let receiver_id = cv.get_receiver();
let user_id = cv.get_sender();
// Authority check: user must be linked to this Iota
if !self.get_user_ids().await.contains(&user_id) {
log_err!(
self.iota_id as i64,
PrintType::Iota,
"Rejected get_chats: sender_id={} is not authorized for this iota.",
user_id
);
return;
}
let mut interested_ids: Vec<i64> = Vec::new();
// ============================
// Load Calls
// ============================
let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(receiver_id).await;
let calls: Vec<Arc<CallGroup>> = call_manager::get_call_groups(user_id).await;
let mut invites: HashMap<i64, Vec<DataValue>> = HashMap::new();
let mut global_calls: Vec<DataValue> = Vec::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 members = call.members.read().await;
let inviter_id = inviter.user_id;
// Find ourselves in the call to get our specific status
let call_self = members.iter().find(|m| m.user_id == user_id).cloned();
if let Some(call_self) = call_self {
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();
// List of all members in the call
let member_ids: Vec<DataValue> = members
.iter()
.map(|m| DataValue::Number(m.user_id as i64))
.collect();
call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string()));
// Build base call container
let mut base_call_map: BTreeMap<DataTypes, DataValue> = BTreeMap::new();
base_call_map.insert(DataTypes::call_id, DataValue::Str(call.call_id.to_string()));
base_call_map.insert(DataTypes::call_members, DataValue::Array(member_ids));
if timeout > 0 {
call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64));
base_call_map.insert(DataTypes::timeout, DataValue::Number(timeout as i64));
}
if admin {
call_map.insert(DataTypes::has_admin, DataValue::Bool(true));
base_call_map.insert(DataTypes::has_admin, DataValue::Bool(true));
}
let call_container = DataValue::container_from_map(&call_map);
// Add to global calls (without contact-specific secret)
global_calls.push(DataValue::container_from_map(&base_call_map));
invites
.entry(inviter_id as i64)
.or_insert_with(Vec::new)
.push(call_container);
// Attach this call to EVERY member of the call (other than ourselves)
for member in members.iter() {
let member_id = member.user_id;
if member_id == user_id {
continue;
}
let mut contact_call_map = base_call_map.clone();
// Add secret if it exists for this pairing
if let Some(secret) = call.secrets.read().await.get(&(member_id, user_id)) {
contact_call_map
.insert(DataTypes::call_secret, DataValue::Str(secret.clone()));
}
invites
.entry(member_id as i64)
.or_insert_with(Vec::new)
.push(DataValue::container_from_map(&contact_call_map));
}
}
}
@ -374,13 +412,10 @@ impl IotaConnection {
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);
if let Some(DataValue::Number(id)) = user_map.get(&DataTypes::user_id) {
interested_ids.push(*id);
// attach calls if exists
if let Some(call_list) = invites.get(user_id)
if let Some(call_list) = invites.get(id)
&& !call_list.is_empty()
{
user_map
@ -399,22 +434,25 @@ impl IotaConnection {
// ============================
// Notify Omega
// ============================
OmegaConnection::user_states(receiver_id as i64, interested_ids.clone()).await;
OmegaConnection::user_states(user_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)
.set_interested(user_id as i64, interested_ids)
.await;
}
// ============================
// Forward to client
// ============================
self.forward_to_client(cv.add_data(DataTypes::user_ids, enriched_contacts))
.await;
self.forward_to_client(
cv.add_data(DataTypes::user_ids, enriched_contacts)
.add_data(DataTypes::calls, DataValue::Array(global_calls)),
)
.await;
}
/// Forward message to client

1
src/rho/rho_connection.rs
View file

@ -77,6 +77,7 @@ impl RhoConnection {
collections
}
#[allow(unused)]
pub async fn get_app_connections(
&self,
userid: Option<i64>,