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

Initial MTP Migration [Broken]

Browse source
This commit is contained in:
Alex Emmet 2026-06-28 13:44:58 +02:00
commit b2a22456ff
30 changed files with 1886 additions and 1574 deletions
Download patch file Download diff file Expand all files Collapse all files

3
communities/Cargo.toml
View file

@ -4,8 +4,7 @@ version = "0.1.0"
edition = "2024"
[dependencies]
ttp-core = { git = "https://git.methanium.net/Tensamin/TTP.git", package = "ttp-core" }
ttp-native = { git = "https://git.methanium.net/Tensamin/TTP.git", package = "ttp-native" }
mtp = { git = "https://git.methanium.net/Methanium/mtp.git" }
iota-logger = { path = "../iota-logger" }
iota-util = { path = "../iota-util" }
iota-storage = { path = "../iota-storage" }

6
communities/src/community.rs
View file

@ -210,7 +210,7 @@ impl Community {
for interactable in target_interactables.iter() {
if interactable.get_name() == name {
if interactable.get_codec() == "category" {
return CommunicationValue::new(CommunicationType::error_internal);
return CommunicationValue::new(CommunicationType::ErrorInternal);
} else {
// cannot move a value of type dyn Interactable the size of dyn Interactable cannot be statically determined (rustc E0161)
return interactable.run_function(cv.clone()).await;
@ -231,12 +231,12 @@ impl Community {
.run_function(cv.clone())
.await;
} else {
return CommunicationValue::new(CommunicationType::error_internal);
return CommunicationValue::new(CommunicationType::ErrorInternal);
}
}
}
}
CommunicationValue::new(CommunicationType::add_conversation)
CommunicationValue::new(CommunicationType::AddConversation)
}
pub async fn save(&self) {

72
communities/src/community_connection.rs
View file

@ -70,12 +70,12 @@ impl CommunityConnection {
let user_id = self.get_user_id().await;
cv = cv.with_sender(user_id);
if cv.is_type(CommunicationType::identification) && !self.is_identified().await {
if cv.is_type(CommunicationType::Identification) && !self.is_identified().await {
self.handle_identification(cv).await;
return;
}
if cv.is_type(CommunicationType::challenge_response) && !self.is_identified().await {
if cv.is_type(CommunicationType::ChallengeResponse) && !self.is_identified().await {
self.handle_challenge_response(cv).await;
return;
}
@ -84,25 +84,25 @@ impl CommunityConnection {
return;
}
if cv.is_type(CommunicationType::ping) {
if cv.is_type(CommunicationType::Ping) {
self.handle_ping(cv).await;
return;
}
if cv.is_type(CommunicationType::client_changed) {
if cv.is_type(CommunicationType::ClientChanged) {
//self.handle_client_changed(cv).await;
return;
}
if cv.is_type(CommunicationType::function) {
if cv.is_type(CommunicationType::Function) {
self.handle_function(cv).await;
return;
}
}
async fn handle_function(&self, cv: CommunicationValue) {
let name = cv.get_data(DataTypes::name).unwrap().as_str().unwrap();
let path = cv.get_data(DataTypes::path).unwrap().as_str().unwrap();
let function = cv.get_data(DataTypes::function).unwrap().as_str().unwrap();
let name = cv.get_data(DataType::Name).unwrap().as_str().unwrap();
let path = cv.get_data(DataType::Path).unwrap().as_str().unwrap();
let function = cv.get_data(DataType::Function).unwrap().as_str().unwrap();
let result = self
.get_community()
@ -115,13 +115,13 @@ impl CommunityConnection {
}
async fn handle_identification(&self, cv: CommunicationValue) {
let user_id = cv
.get_data(DataTypes::user_id)
.get_data(DataType::UserId)
.unwrap_or(&JsonValue::Number(Number::from(0)))
.as_i64()
.unwrap_or(0);
let Some(user) = get_user(user_id) else {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_user_id)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
return;
};
@ -151,7 +151,7 @@ impl CommunityConnection {
let user_public_key_bytes = match STANDARD.decode(&user.public_key) {
Ok(bytes) => bytes,
Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_user_id)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
return;
}
@ -160,14 +160,14 @@ impl CommunityConnection {
let user_pub_key: PublicKey = match PublicKey::from_bytes(&user_public_key_bytes) {
Some(key) => key,
__ => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_user_id)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
return;
}
};
let Some(community) = self.community.read().await.clone() else {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
};
@ -178,7 +178,7 @@ impl CommunityConnection {
let shared_secret = match community_private_key.to_diffie_hellman(&user_pub_key) {
Some(secret) => secret,
_ => {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
}
@ -200,7 +200,7 @@ impl CommunityConnection {
let encrypted_challenge = match cipher.encrypt(nonce, challenge_str.as_bytes()) {
Ok(data) => data,
Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
}
@ -209,21 +209,21 @@ impl CommunityConnection {
let mut encrypted_out = nonce_bytes.to_vec();
encrypted_out.extend(encrypted_challenge);
let response = CommunicationValue::new(CommunicationType::challenge)
let response = CommunicationValue::new(CommunicationType::Challenge)
.add_data_str(
DataTypes::public_key,
DataType::PublicKey,
STANDARD.encode(community_public_key.as_bytes()),
)
.add_data_str(DataTypes::challenge, STANDARD.encode(&encrypted_out))
.add_data_str(DataType::Challenge, STANDARD.encode(&encrypted_out))
.with_id(cv.get_id());
self.send_message(&response).await;
}
async fn handle_challenge_response(self: Arc<Self>, cv: CommunicationValue) {
let client_challenge_response_b64 = match cv.get_data(DataTypes::challenge) {
let client_challenge_response_b64 = match cv.get_data(DataType::Challenge) {
Some(data) => data.to_string(),
_ => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData)
.await;
return;
}
@ -232,38 +232,38 @@ impl CommunityConnection {
let challenge_response_bytes = match STANDARD.decode(&client_challenge_response_b64) {
Ok(bytes) => bytes,
Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData)
.await;
return;
}
};
if challenge_response_bytes.len() < 12 {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData)
.await;
return;
}
let Some(user) = self.auth.read().await.clone() else {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
};
let Some(user_pub_bytes) = STANDARD.decode(&user.public_key).ok() else {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData)
.await;
return;
};
let Some(user_pub_key) = PublicKey::from_bytes(&user_pub_bytes) else {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_public_key)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidPublicKey)
.await;
return;
};
let Some(community) = self.community.read().await.clone() else {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
};
@ -273,7 +273,7 @@ impl CommunityConnection {
let shared_secret = match community_private_key.to_diffie_hellman(&user_pub_key) {
Some(secret) => secret,
_ => {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
}
@ -297,7 +297,7 @@ impl CommunityConnection {
let decrypted_bytes = match cipher.decrypt(nonce, ciphertext) {
Ok(pt) => pt,
Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_challenge)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
return;
}
@ -306,7 +306,7 @@ impl CommunityConnection {
let client_response = match String::from_utf8(decrypted_bytes) {
Ok(str) => str,
Err(_) => {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_data)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidData)
.await;
return;
}
@ -315,7 +315,7 @@ impl CommunityConnection {
let expected_challenge = self.challenge.read().await.clone();
if client_response != expected_challenge {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_challenge)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidChallenge)
.await;
self.close().await;
return;
@ -327,7 +327,7 @@ impl CommunityConnection {
}
let Some(community) = self.community.read().await.clone() else {
self.send_error_response(&cv.get_id(), CommunicationType::error_internal)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInternal)
.await;
return;
};
@ -335,15 +335,15 @@ impl CommunityConnection {
let user_id = self.get_user_id().await;
if user_id == 0 {
self.send_error_response(&cv.get_id(), CommunicationType::error_invalid_user_id)
self.send_error_response(&cv.get_id(), CommunicationType::ErrorInvalidUserId)
.await;
return;
}
arc.add_connection(self.clone()).await;
let response = CommunicationValue::new(CommunicationType::identification_response)
.add_data(DataTypes::interactables, {
let response = CommunicationValue::new(CommunicationType::IdentificationResponse)
.add_data(DataType::Interactables, {
let a: Vec<Arc<Box<dyn Interactable>>> = arc.get_interactables(user_id).await;
let mut c: JsonValue = JsonValue::new_object();
for b in a {
@ -382,14 +382,14 @@ impl CommunityConnection {
}
async fn handle_ping(&self, cv: CommunicationValue) {
if let Some(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Some(last_ping) = cv.get_data(DataType::LastPing) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await;
*ping_guard = ping_val;
}
}
let response = CommunicationValue::new(CommunicationType::pong).with_id(cv.get_id());
let response = CommunicationValue::new(CommunicationType::Pong).with_id(cv.get_id());
self.send_message(&response).await;
}

242
communities/src/interactables/category.rs
View file

@ -1,121 +1,121 @@
use crate::communities::{community::Community, interactables::interactable::Interactable};
use async_trait::async_trait;
use json::JsonValue;
use std::any::Any;
use std::sync::Arc;
use ttp_core::CommunicationValue;
use uuid::Uuid;
pub struct Category {
id: Uuid,
name: String,
path: String,
community: Arc<Community>,
children: Vec<Arc<Box<dyn Interactable>>>,
}
impl Category {
pub fn new() -> Category {
Category {
id: Uuid::new_v4(),
name: String::new(),
path: String::new(),
community: Arc::new(Community::new()),
children: Vec::new(),
}
}
pub fn get_child(&self, path: String, name: String) -> Option<Arc<Box<dyn Interactable>>> {
if path.is_empty() {
self.children
.iter()
.find(|child| child.get_name() == &name)
.cloned()
} else {
let sub_module = path.split("/").next().unwrap();
let next = self
.children
.iter()
.find(|child| child.get_name() == sub_module)
.unwrap();
if next.get_codec() == "category" {
let next_cat = next.as_any().downcast_ref::<Category>().unwrap();
next_cat.get_child(path, name)
} else {
Some(next.clone())
}
}
}
pub fn get_children(&self) -> Vec<Arc<Box<dyn Interactable>>> {
self.children.iter().map(|child| child.clone()).collect()
}
}
#[async_trait]
impl Interactable for Category {
fn get_id(&self) -> &Uuid {
&self.id
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn get_codec(&self) -> String {
"category".to_string()
}
fn set_name(&mut self, name: String) {
self.name = name;
}
fn set_path(&mut self, path: String) {
self.path = path;
}
fn get_community(&self) -> &Arc<Community> {
&self.community
}
fn set_community(&mut self, community: Arc<Community>) {
self.community = community;
}
fn get_name(&self) -> &String {
&self.name
}
fn get_path(&self) -> &String {
&self.path
}
fn get_total_path(&self) -> String {
String::new() + &self.path + "/" + &self.name
}
fn get_data(&self) -> JsonValue {
let mut v = JsonValue::new_object();
for child in &self.children {
let mut subject = JsonValue::new_object();
subject["codec"] = JsonValue::String(child.get_codec());
subject["data"] = child.get_data();
v[child.get_name()] = subject;
}
v
}
async fn run_function(&self, _cv: CommunicationValue) -> CommunicationValue {
CommunicationValue::new(CommunicationType::error_internal)
}
fn to_json(&self) -> JsonValue {
let mut v = JsonValue::new_object();
v["children"] = JsonValue::new_array();
for child in &self.children {
let _ = v["children"].push(child.to_json());
}
v
}
fn load(
&mut self,
community: Arc<Community>,
id: Uuid,
path: String,
name: String,
_json: &JsonValue,
) {
self.community = community;
self.id = id;
self.name = name;
self.path = path;
}
}
use crate::communities::{community::Community, interactables::interactable::Interactable};
use async_trait::async_trait;
use json::JsonValue;
use std::any::Any;
use std::sync::Arc;
use mtp::codec::CommunicationValue;
use uuid::Uuid;
pub struct Category {
id: Uuid,
name: String,
path: String,
community: Arc<Community>,
children: Vec<Arc<Box<dyn Interactable>>>,
}
impl Category {
pub fn new() -> Category {
Category {
id: Uuid::new_v4(),
name: String::new(),
path: String::new(),
community: Arc::new(Community::new()),
children: Vec::new(),
}
}
pub fn get_child(&self, path: String, name: String) -> Option<Arc<Box<dyn Interactable>>> {
if path.is_empty() {
self.children
.iter()
.find(|child| child.get_name() == &name)
.cloned()
} else {
let sub_module = path.split("/").next().unwrap();
let next = self
.children
.iter()
.find(|child| child.get_name() == sub_module)
.unwrap();
if next.get_codec() == "category" {
let next_cat = next.as_any().downcast_ref::<Category>().unwrap();
next_cat.get_child(path, name)
} else {
Some(next.clone())
}
}
}
pub fn get_children(&self) -> Vec<Arc<Box<dyn Interactable>>> {
self.children.iter().map(|child| child.clone()).collect()
}
}
#[async_trait]
impl Interactable for Category {
fn get_id(&self) -> &Uuid {
&self.id
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn get_codec(&self) -> String {
"category".to_string()
}
fn set_name(&mut self, name: String) {
self.name = name;
}
fn set_path(&mut self, path: String) {
self.path = path;
}
fn get_community(&self) -> &Arc<Community> {
&self.community
}
fn set_community(&mut self, community: Arc<Community>) {
self.community = community;
}
fn get_name(&self) -> &String {
&self.name
}
fn get_path(&self) -> &String {
&self.path
}
fn get_total_path(&self) -> String {
String::new() + &self.path + "/" + &self.name
}
fn get_data(&self) -> JsonValue {
let mut v = JsonValue::new_object();
for child in &self.children {
let mut subject = JsonValue::new_object();
subject["codec"] = JsonValue::String(child.get_codec());
subject["data"] = child.get_data();
v[child.get_name()] = subject;
}
v
}
async fn run_function(&self, _cv: CommunicationValue) -> CommunicationValue {
CommunicationValue::new(CommunicationType::ErrorInternal)
}
fn to_json(&self) -> JsonValue {
let mut v = JsonValue::new_object();
v["children"] = JsonValue::new_array();
for child in &self.children {
let _ = v["children"].push(child.to_json());
}
v
}
fn load(
&mut self,
community: Arc<Community>,
id: Uuid,
path: String,
name: String,
_json: &JsonValue,
) {
self.community = community;
self.id = id;
self.name = name;
self.path = path;
}
}

2
communities/src/interactables/interactable.rs
View file

@ -3,7 +3,7 @@ use async_trait::async_trait;
use json::JsonValue;
use std::any::Any;
use std::sync::Arc;
use ttp_core::CommunicationValue;
use mtp::codec::CommunicationValue;
use uuid::Uuid;
pub type InteractableFactory = fn() -> Box<dyn Interactable>;

528
communities/src/interactables/text_chat.rs
View file

@ -1,264 +1,264 @@
use crate::{
communities::{
community::Community, community_connection::CommunityConnection,
interactables::interactable::Interactable,
},
log,
util::file_util::{get_children, load_file, save_file},
};
use async_trait::async_trait;
use json::{JsonValue, array, object};
use std::fs;
use std::path::Path;
use std::sync::Arc;
use std::{any::Any, collections::HashMap};
use ttp_core::{CommunicationType, CommunicationValue, DataTypes};
use uuid::Uuid;
pub struct TextChat {
id: Uuid,
name: String,
path: String,
community: Arc<Community>,
}
impl TextChat {
pub fn new() -> TextChat {
TextChat {
id: Uuid::new_v4(),
name: String::new(),
path: String::new(),
community: Arc::new(Community::new()),
}
}
pub fn add_message(&self, send_time: u128, sender: i64, message: &str) {
let user_dir = &format!(
"communities/{}/interactables/{}/{}",
self.get_community().get_name(),
self.get_path(),
self.get_name()
);
let working_dir = iota_util::file_util::get_directory();
let full_dir = Path::new(&working_dir).join(user_dir);
if let Err(e) = fs::create_dir_all(&full_dir) {
log!("Failed to create chat directory: {}", e);
return;
}
let mut chunk_index = 0;
let mut message_chunk = array![];
// find latest chunk not full (max 800 msgs)
loop {
let file_name = format!("msgs_{}.json", chunk_index);
let file_content = load_file(&user_dir, &file_name);
if !file_content.is_empty() {
if let Ok(current_chunk) = json::parse(&file_content) {
if current_chunk.is_array() && current_chunk.len() < 800 {
message_chunk = current_chunk;
break;
}
} else {
log!("Failed to parse existing JSON file: {}", file_name);
}
} else {
break;
}
chunk_index += 1;
if chunk_index > 1000 {
log!("Too many message chunks. Aborting add.");
return;
}
}
let json_obj = object! {
"timestamp" => send_time as i64,
"content" => message,
"sender" => sender.to_string(),
};
if let Err(e) = message_chunk.push(json_obj) {
log!("Failed to push new message into JSON array: {}", e);
return;
}
let file_name = format!("msgs_{}.json", chunk_index);
log!("Saving message to {}/{}", user_dir, file_name);
save_file(&user_dir, &file_name, &message_chunk.dump());
}
pub fn get_messages(&self, loaded_messages: i64, amount: i64) -> JsonValue {
let mut messages = array![];
let mut latest_chunk_index: i32 = -1;
let files = get_children(&format!(
"communities/{}/interactables/{}/{}",
self.get_community().get_name(),
self.get_path(),
self.get_name()
));
for entry in files {
if let Some(num) = {
entry
.strip_prefix("msgs_")
.and_then(|s| s.strip_suffix(".json"))
} {
if let Ok(index) = num.parse::<i32>() {
if index > latest_chunk_index {
latest_chunk_index = index;
}
}
}
}
if latest_chunk_index == -1 {
return messages;
}
let mut to_skip = loaded_messages;
let mut needed = amount;
for chunk_index in (0..=latest_chunk_index).rev() {
if needed == 0 {
break;
}
let file_name = format!("msgs_{}.json", chunk_index);
let file_content = load_file(
&format!(
"communities/{}/interactables/{}/{}",
self.get_community().get_name(),
self.get_path(),
self.get_name()
),
&file_name,
);
if file_content.is_empty() {
continue;
}
if let Ok(chunk) = json::parse(&file_content) {
for i in (0..chunk.len()).rev() {
if needed == 0 {
break;
}
if to_skip > 0 {
to_skip -= 1;
continue;
}
messages.push(chunk[i].clone()).unwrap();
needed -= 1;
}
}
}
messages
}
}
#[async_trait]
impl Interactable for TextChat {
fn get_id(&self) -> &Uuid {
&self.id
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn get_codec(&self) -> String {
"text".to_string()
}
fn set_name(&mut self, name: String) {
self.name = name;
}
fn set_path(&mut self, path: String) {
self.path = path;
}
fn get_community(&self) -> &Arc<Community> {
&self.community
}
fn set_community(&mut self, community: Arc<Community>) {
self.community = community;
}
fn get_name(&self) -> &String {
&self.name
}
fn get_path(&self) -> &String {
&self.path
}
fn get_total_path(&self) -> String {
String::new() + &self.path + "/" + &self.name
}
fn get_data(&self) -> JsonValue {
JsonValue::new_object()
}
async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
let payload = cv.get_data(DataTypes::payload).as_container().unwrap();
if cv.get_data(DataTypes::function).as_str().unwrap() == "get_messages" {
let amount = payload.get(DataTypes::amount).as_i64().unwrap();
let loaded_messages = payload["loaded_messages"].as_i64().unwrap();
let messages = self.get_messages(loaded_messages, amount).clone();
let mut payload = JsonValue::new_object();
payload["messages"] = messages;
return CommunicationValue::new(CommunicationType::function)
.with_id(cv.get_id())
.add_data_str(DataTypes::name, self.name.clone())
.add_data_str(DataTypes::path, self.path.clone())
.add_data_str(DataTypes::result, "message_chunk".to_string())
.add_data(DataTypes::payload, payload);
}
if cv.get_data(DataTypes::function).unwrap().as_str().unwrap() == "send_message" {
let message = payload["message"].as_str().unwrap();
let milliseconds_timestamp: u128 = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis();
self.add_message(milliseconds_timestamp, cv.get_sender(), message);
let mut distribution_payload = JsonValue::new_object();
distribution_payload["message"] = JsonValue::String(message.to_string());
distribution_payload["sender_id"] = JsonValue::String(cv.get_sender().to_string());
distribution_payload["send_time"] =
JsonValue::String(milliseconds_timestamp.to_string());
let distribution = CommunicationValue::new(CommunicationType::update)
.with_id(cv.get_id())
.add_data_str(DataTypes::name, self.name.clone())
.add_data_str(DataTypes::path, self.path.clone())
.add_data_str(DataTypes::result, "message_live".to_string())
.add_data(DataTypes::payload, distribution_payload);
let connections: HashMap<i64, Vec<Arc<CommunityConnection>>> =
self.get_community().get_connections().await.clone();
for con in connections.values() {
for c in con {
let cd: &Arc<CommunityConnection> = c;
cd.send_message(&distribution).await;
}
}
return CommunicationValue::new(CommunicationType::function)
.with_id(cv.get_id())
.add_data_str(DataTypes::name, self.name.clone())
.add_data_str(DataTypes::path, self.path.clone())
.add_data_str(DataTypes::result, "message_received".to_string())
.add_data(DataTypes::payload, JsonValue::new_object());
}
CommunicationValue::new(CommunicationType::error_internal).with_id(cv.get_id())
}
fn to_json(&self) -> JsonValue {
JsonValue::new_object()
}
fn load(
&mut self,
community: Arc<Community>,
id: Uuid,
path: String,
name: String,
_json: &JsonValue,
) {
self.community = community;
self.id = id;
self.name = name;
self.path = path;
}
}
use crate::{
communities::{
community::Community, community_connection::CommunityConnection,
interactables::interactable::Interactable,
},
log,
util::file_util::{get_children, load_file, save_file},
};
use async_trait::async_trait;
use json::{JsonValue, array, object};
use std::fs;
use std::path::Path;
use std::sync::Arc;
use std::{any::Any, collections::HashMap};
use mtp::codec::{CommunicationType, CommunicationValue, DataType};
use uuid::Uuid;
pub struct TextChat {
id: Uuid,
name: String,
path: String,
community: Arc<Community>,
}
impl TextChat {
pub fn new() -> TextChat {
TextChat {
id: Uuid::new_v4(),
name: String::new(),
path: String::new(),
community: Arc::new(Community::new()),
}
}
pub fn add_message(&self, send_time: u128, sender: i64, message: &str) {
let user_dir = &format!(
"communities/{}/interactables/{}/{}",
self.get_community().get_name(),
self.get_path(),
self.get_name()
);
let working_dir = iota_util::file_util::get_directory();
let full_dir = Path::new(&working_dir).join(user_dir);
if let Err(e) = fs::create_dir_all(&full_dir) {
log!("Failed to create chat directory: {}", e);
return;
}
let mut chunk_index = 0;
let mut message_chunk = array![];
// find latest chunk not full (max 800 msgs)
loop {
let file_name = format!("msgs_{}.json", chunk_index);
let file_content = load_file(&user_dir, &file_name);
if !file_content.is_empty() {
if let Ok(current_chunk) = json::parse(&file_content) {
if current_chunk.is_array() && current_chunk.len() < 800 {
message_chunk = current_chunk;
break;
}
} else {
log!("Failed to parse existing JSON file: {}", file_name);
}
} else {
break;
}
chunk_index += 1;
if chunk_index > 1000 {
log!("Too many message chunks. Aborting add.");
return;
}
}
let json_obj = object! {
"timestamp" => send_time as i64,
"content" => message,
"sender" => sender.to_string(),
};
if let Err(e) = message_chunk.push(json_obj) {
log!("Failed to push new message into JSON array: {}", e);
return;
}
let file_name = format!("msgs_{}.json", chunk_index);
log!("Saving message to {}/{}", user_dir, file_name);
save_file(&user_dir, &file_name, &message_chunk.dump());
}
pub fn get_messages(&self, loaded_messages: i64, amount: i64) -> JsonValue {
let mut messages = array![];
let mut latest_chunk_index: i32 = -1;
let files = get_children(&format!(
"communities/{}/interactables/{}/{}",
self.get_community().get_name(),
self.get_path(),
self.get_name()
));
for entry in files {
if let Some(num) = {
entry
.strip_prefix("msgs_")
.and_then(|s| s.strip_suffix(".json"))
} {
if let Ok(index) = num.parse::<i32>() {
if index > latest_chunk_index {
latest_chunk_index = index;
}
}
}
}
if latest_chunk_index == -1 {
return messages;
}
let mut to_skip = loaded_messages;
let mut needed = amount;
for chunk_index in (0..=latest_chunk_index).rev() {
if needed == 0 {
break;
}
let file_name = format!("msgs_{}.json", chunk_index);
let file_content = load_file(
&format!(
"communities/{}/interactables/{}/{}",
self.get_community().get_name(),
self.get_path(),
self.get_name()
),
&file_name,
);
if file_content.is_empty() {
continue;
}
if let Ok(chunk) = json::parse(&file_content) {
for i in (0..chunk.len()).rev() {
if needed == 0 {
break;
}
if to_skip > 0 {
to_skip -= 1;
continue;
}
messages.push(chunk[i].clone()).unwrap();
needed -= 1;
}
}
}
messages
}
}
#[async_trait]
impl Interactable for TextChat {
fn get_id(&self) -> &Uuid {
&self.id
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn get_codec(&self) -> String {
"text".to_string()
}
fn set_name(&mut self, name: String) {
self.name = name;
}
fn set_path(&mut self, path: String) {
self.path = path;
}
fn get_community(&self) -> &Arc<Community> {
&self.community
}
fn set_community(&mut self, community: Arc<Community>) {
self.community = community;
}
fn get_name(&self) -> &String {
&self.name
}
fn get_path(&self) -> &String {
&self.path
}
fn get_total_path(&self) -> String {
String::new() + &self.path + "/" + &self.name
}
fn get_data(&self) -> JsonValue {
JsonValue::new_object()
}
async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
let payload = cv.get_data(DataType::Payload).as_container().unwrap();
if cv.get_data(DataType::Function).as_str().unwrap() == "get_messages" {
let amount = payload.get(DataType::Amount).as_i64().unwrap();
let loaded_messages = payload["loaded_messages"].as_i64().unwrap();
let messages = self.get_messages(loaded_messages, amount).clone();
let mut payload = JsonValue::new_object();
payload["messages"] = messages;
return CommunicationValue::new(CommunicationType::Function)
.with_id(cv.get_id())
.add_data_str(DataType::Name, self.name.clone())
.add_data_str(DataType::Path, self.path.clone())
.add_data_str(DataType::Result, "message_chunk".to_string())
.add_data(DataType::Payload, payload);
}
if cv.get_data(DataType::Function).unwrap().as_str().unwrap() == "send_message" {
let message = payload["message"].as_str().unwrap();
let milliseconds_timestamp: u128 = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis();
self.add_message(milliseconds_timestamp, cv.get_sender(), message);
let mut distribution_payload = JsonValue::new_object();
distribution_payload["message"] = JsonValue::String(message.to_string());
distribution_payload["sender_id"] = JsonValue::String(cv.get_sender().to_string());
distribution_payload["send_time"] =
JsonValue::String(milliseconds_timestamp.to_string());
let distribution = CommunicationValue::new(CommunicationType::Update)
.with_id(cv.get_id())
.add_data_str(DataType::Name, self.name.clone())
.add_data_str(DataType::Path, self.path.clone())
.add_data_str(DataType::Result, "message_live".to_string())
.add_data(DataType::Payload, distribution_payload);
let connections: HashMap<i64, Vec<Arc<CommunityConnection>>> =
self.get_community().get_connections().await.clone();
for con in connections.values() {
for c in con {
let cd: &Arc<CommunityConnection> = c;
cd.send_message(&distribution).await;
}
}
return CommunicationValue::new(CommunicationType::Function)
.with_id(cv.get_id())
.add_data_str(DataType::Name, self.name.clone())
.add_data_str(DataType::Path, self.path.clone())
.add_data_str(DataType::Result, "message_received".to_string())
.add_data(DataType::Payload, JsonValue::new_object());
}
CommunicationValue::new(CommunicationType::ErrorInternal).with_id(cv.get_id())
}
fn to_json(&self) -> JsonValue {
JsonValue::new_object()
}
fn load(
&mut self,
community: Arc<Community>,
id: Uuid,
path: String,
name: String,
_json: &JsonValue,
) {
self.community = community;
self.id = id;
self.name = name;
self.path = path;
}
}

374
communities/src/interactables/voice_chat.rs
View file

@ -1,187 +1,187 @@
use crate::communities::{community::Community, interactables::interactable::Interactable};
use async_trait::async_trait;
use json::JsonValue;
use std::sync::Arc;
use std::{any::Any, sync::RwLock};
use uuid::Uuid;
pub enum CallUserState {
Active,
Muted,
Deafed,
}
impl CallUserState {
pub fn parse(state: &str) -> CallUserState {
match state {
"active" => CallUserState::Active,
"muted" => CallUserState::Muted,
"deafed" => CallUserState::Deafed,
_ => CallUserState::Active,
}
}
pub fn to_string(&self) -> String {
match self {
CallUserState::Active => "active".to_string(),
CallUserState::Muted => "muted".to_string(),
CallUserState::Deafed => "deafed".to_string(),
}
}
}
pub struct CallUser {
pub user_id: Uuid,
pub user_state: CallUserState,
pub streaming: bool,
}
pub struct VoiceChat {
id: Uuid,
name: String,
path: String,
community: Arc<Community>,
users: RwLock<Vec<CallUser>>,
}
impl VoiceChat {
pub fn new() -> VoiceChat {
VoiceChat {
id: Uuid::new_v4(),
name: String::new(),
path: String::new(),
community: Arc::new(Community::new()),
users: RwLock::new(Vec::new()),
}
}
pub fn update_user_state(
self: Arc<Self>,
user_id: Uuid,
state: CallUserState,
streaming: bool,
) {
if let Some(user) = self
.users
.write()
.unwrap()
.iter_mut()
.find(|u| u.user_id == user_id)
{
user.user_state = state;
user.streaming = streaming;
}
}
}
#[async_trait]
impl Interactable for VoiceChat {
fn get_id(&self) -> &Uuid {
&self.id
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn get_codec(&self) -> String {
"voice".to_string()
}
fn set_name(&mut self, name: String) {
self.name = name;
}
fn set_path(&mut self, path: String) {
self.path = path;
}
fn get_community(&self) -> &Arc<Community> {
&self.community
}
fn set_community(&mut self, community: Arc<Community>) {
self.community = community;
}
fn get_name(&self) -> &String {
&self.name
}
fn get_path(&self) -> &String {
&self.path
}
fn get_total_path(&self) -> String {
String::new() + &self.path + "/" + &self.name
}
fn get_data(&self) -> JsonValue {
let mut data = JsonValue::new_object();
let mut active_users = JsonValue::new_object();
for user in self.users.read().unwrap().iter() {
let mut user_data = JsonValue::new_object();
let _ = user_data.insert("state", JsonValue::String(user.user_state.to_string()));
let _ = user_data.insert("streaming", JsonValue::Boolean(user.streaming));
let _ = active_users.insert(&user.user_id.to_string(), user_data);
}
let _ = data.insert("active_users", active_users);
data
}
async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
let payload = cv.get_data(DataTypes::payload).unwrap();
let function = cv.get_data(DataTypes::function).unwrap().as_str().unwrap();
if function == "get_call" {
let sender_id = payload["sender_id"].as_str().unwrap();
let message_id = payload["message"].as_str().unwrap();
let send_time = payload["send_time"].as_str().unwrap();
let mut response_payload = JsonValue::new_object();
response_payload["sender_id"] = JsonValue::String(sender_id.to_string());
response_payload["message"] = JsonValue::String(message_id.to_string());
response_payload["send_time"] = JsonValue::String(send_time.to_string());
return CommunicationValue::new(CommunicationType::function)
.with_id(cv.get_id())
.add_data_str(DataTypes::name, self.name.clone())
.add_data_str(DataTypes::path, self.path.clone())
.add_data_str(DataTypes::result, "getting_call".to_string())
.add_data(DataTypes::payload, response_payload);
}
if function == "update_user_state" {
let user_id = payload["user_id"].as_str().unwrap();
let state = payload["state"].as_str().unwrap();
let streaming = payload["streaming"].as_bool().unwrap();
if let Some(user) = self
.users
.write()
.unwrap()
.iter_mut()
.find(|u| u.user_id == Uuid::parse_str(user_id).unwrap())
{
user.user_state = CallUserState::parse(state);
user.streaming = streaming;
}
let mut response_payload = JsonValue::new_object();
response_payload["user_id"] = JsonValue::String(user_id.to_string());
response_payload["state"] = JsonValue::String(state.to_string());
response_payload["streaming"] = JsonValue::Boolean(streaming);
return CommunicationValue::new(CommunicationType::update)
.with_id(cv.get_id())
.add_data_str(DataTypes::name, self.name.clone())
.add_data_str(DataTypes::path, self.path.clone())
.add_data_str(DataTypes::result, "user_changed".to_string())
.add_data(DataTypes::payload, response_payload);
}
CommunicationValue::new(CommunicationType::error_internal).with_id(cv.get_id())
}
fn to_json(&self) -> JsonValue {
let v = JsonValue::new_object();
v
}
fn load(
&mut self,
community: Arc<Community>,
id: Uuid,
path: String,
name: String,
_json: &JsonValue,
) {
self.community = community;
self.id = id;
self.name = name;
self.path = path;
}
}
use crate::communities::{community::Community, interactables::interactable::Interactable};
use async_trait::async_trait;
use json::JsonValue;
use std::sync::Arc;
use std::{any::Any, sync::RwLock};
use uuid::Uuid;
pub enum CallUserState {
Active,
Muted,
Deafed,
}
impl CallUserState {
pub fn parse(state: &str) -> CallUserState {
match state {
"active" => CallUserState::Active,
"muted" => CallUserState::Muted,
"deafed" => CallUserState::Deafed,
_ => CallUserState::Active,
}
}
pub fn to_string(&self) -> String {
match self {
CallUserState::Active => "active".to_string(),
CallUserState::Muted => "muted".to_string(),
CallUserState::Deafed => "deafed".to_string(),
}
}
}
pub struct CallUser {
pub user_id: Uuid,
pub user_state: CallUserState,
pub streaming: bool,
}
pub struct VoiceChat {
id: Uuid,
name: String,
path: String,
community: Arc<Community>,
users: RwLock<Vec<CallUser>>,
}
impl VoiceChat {
pub fn new() -> VoiceChat {
VoiceChat {
id: Uuid::new_v4(),
name: String::new(),
path: String::new(),
community: Arc::new(Community::new()),
users: RwLock::new(Vec::new()),
}
}
pub fn update_user_state(
self: Arc<Self>,
user_id: Uuid,
state: CallUserState,
streaming: bool,
) {
if let Some(user) = self
.users
.write()
.unwrap()
.iter_mut()
.find(|u| u.user_id == user_id)
{
user.user_state = state;
user.streaming = streaming;
}
}
}
#[async_trait]
impl Interactable for VoiceChat {
fn get_id(&self) -> &Uuid {
&self.id
}
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn get_codec(&self) -> String {
"voice".to_string()
}
fn set_name(&mut self, name: String) {
self.name = name;
}
fn set_path(&mut self, path: String) {
self.path = path;
}
fn get_community(&self) -> &Arc<Community> {
&self.community
}
fn set_community(&mut self, community: Arc<Community>) {
self.community = community;
}
fn get_name(&self) -> &String {
&self.name
}
fn get_path(&self) -> &String {
&self.path
}
fn get_total_path(&self) -> String {
String::new() + &self.path + "/" + &self.name
}
fn get_data(&self) -> JsonValue {
let mut data = JsonValue::new_object();
let mut active_users = JsonValue::new_object();
for user in self.users.read().unwrap().iter() {
let mut user_data = JsonValue::new_object();
let _ = user_data.insert("state", JsonValue::String(user.user_state.to_string()));
let _ = user_data.insert("streaming", JsonValue::Boolean(user.streaming));
let _ = active_users.insert(&user.user_id.to_string(), user_data);
}
let _ = data.insert("active_users", active_users);
data
}
async fn run_function(&self, cv: CommunicationValue) -> CommunicationValue {
let payload = cv.get_data(DataType::Payload).unwrap();
let function = cv.get_data(DataType::Function).unwrap().as_str().unwrap();
if function == "get_call" {
let sender_id = payload["sender_id"].as_str().unwrap();
let message_id = payload["message"].as_str().unwrap();
let send_time = payload["send_time"].as_str().unwrap();
let mut response_payload = JsonValue::new_object();
response_payload["sender_id"] = JsonValue::String(sender_id.to_string());
response_payload["message"] = JsonValue::String(message_id.to_string());
response_payload["send_time"] = JsonValue::String(send_time.to_string());
return CommunicationValue::new(CommunicationType::Function)
.with_id(cv.get_id())
.add_data_str(DataType::Name, self.name.clone())
.add_data_str(DataType::Path, self.path.clone())
.add_data_str(DataType::Result, "getting_call".to_string())
.add_data(DataType::Payload, response_payload);
}
if function == "update_user_state" {
let user_id = payload["user_id"].as_str().unwrap();
let state = payload["state"].as_str().unwrap();
let streaming = payload["streaming"].as_bool().unwrap();
if let Some(user) = self
.users
.write()
.unwrap()
.iter_mut()
.find(|u| u.user_id == Uuid::parse_str(user_id).unwrap())
{
user.user_state = CallUserState::parse(state);
user.streaming = streaming;
}
let mut response_payload = JsonValue::new_object();
response_payload["user_id"] = JsonValue::Number(user_id);
response_payload["state"] = JsonValue::String(state.to_string());
response_payload["streaming"] = JsonValue::Boolean(streaming);
return CommunicationValue::new(CommunicationType::Update)
.with_id(cv.get_id())
.add_data_str(DataType::Name, self.name.clone())
.add_data_str(DataType::Path, self.path.clone())
.add_data_str(DataType::Result, "user_changed".to_string())
.add_data(DataType::Payload, response_payload);
}
CommunicationValue::new(CommunicationType::ErrorInternal).with_id(cv.get_id())
}
fn to_json(&self) -> JsonValue {
let v = JsonValue::new_object();
v
}
fn load(
&mut self,
community: Arc<Community>,
id: Uuid,
path: String,
name: String,
_json: &JsonValue,
) {
self.community = community;
self.id = id;
self.name = name;
self.path = path;
}
}