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

Compare commits

base: tensamin:f82500ea7dd0013ad3efd7f0b4620e20ee028f2a
Branches Tags
tensamin:main
tensamin:renovate/zip-8.x
tensamin:renovate/rand_core-0.x
tensamin:renovate/rand-0.x
tensamin:renovate/base64-0.x
tensamin:0.1.0-dev-f252724
tensamin:1.0.0
..
compare: tensamin:9b20d159b50d6224b370319fd10b0f21c393027d
Branches Tags
tensamin:renovate/zip-8.x
tensamin:renovate/rand_core-0.x
tensamin:renovate/rand-0.x
tensamin:renovate/base64-0.x
tensamin:main
tensamin:0.1.0-dev-f252724
tensamin:1.0.0
36 changed files with 2755 additions and 3739 deletions
Download patch file Download diff file Expand all files Collapse all files

628
Cargo.lock generated
View file

File diff suppressed because it is too large Load diff

18
Cargo.toml
View file

@ -1,19 +1,3 @@
[workspace]
members = [
"iota-storage",
"iota-connection",
"client",
"iota-auth",
"other-iota",
"iota-updater",
"iota-terms",
"iota-state",
"iota-cli",
"iota-core",
"omikron-connector",
"web-server",
"web-ui",
"iota-logger",
"iota-util",
]
members = ["iota-storage", "client", "iota-auth", "other-iota", "iota-updater", "iota-terms", "iota-state", "iota-cli", "iota-core", "omikron-connector", "web-server", "web-ui", "iota-logger", "iota-util"]
resolver = "3"

1
client/Cargo.toml
View file

@ -5,7 +5,6 @@ edition = "2024"
[dependencies]
mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["client"] }
iota-connection = { path = "../iota-connection" }
iota-logger = { path = "../iota-logger" }
iota-util = { path = "../iota-util" }
iota-storage = { path = "../iota-storage" }

610
client/src/client_connection.rs
View file

@ -1,22 +1,160 @@
use dashmap::DashMap;
use iota_connection::message_common::*;
use iota_connection::message_handlers;
use iota_logger::{log_cv_in, log_cv_out, log_t};
use iota_state::SHUTDOWN;
use iota_storage::util::chat_files::{self, MessageState, change_message_state};
use iota_storage::users::contact::Contact;
use iota_storage::util::chat_files::{MessageState, change_message_state};
use iota_storage::util::chats_util::{get_user, mod_user};
use iota_storage::util::communities_util::CommunitiesUtil;
use iota_storage::util::config_util::CONFIG;
use iota_storage::util::e2ee_storage::{self, StoredChatSecret};
use iota_storage::util::e2ee_storage::{self, ChatSecretQuery, StoredChatSecret};
use iota_storage::util::{chat_files, chats_util};
use iota_util::crypto_helper::keyring_from_base64;
use iota_util::crypto_util::{self};
use iota_util::file_util::{get_children, load_file, save_file};
use mtp::client::{Receiver, Sender};
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::type_map::TypeMap;
use std::sync::Arc;
use std::time::Duration;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{Mutex, RwLock, mpsc, watch};
use tokio::task::JoinHandle;
use uuid::Uuid;
fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue {
use mtp::type_map::{DataTypeId, TypeMap};
let tm = TypeMap::latest();
DataValue::Container(
items
.into_iter()
.filter_map(|(dt, dv)| tm.data_id_enum(dt).map(|id| (DataTypeId(id), dv)))
.collect(),
)
}
fn data_string(cv: &CommunicationValue, dt: DataType) -> Option<String> {
cv.get_data(dt)
.as_str()
.map(|s| s.to_string())
.or_else(|| cv.get_data(dt).as_number().map(|n| n.to_string()))
.or_else(|| cv.get_data(dt).as_signed_number().map(|n| n.to_string()))
}
fn data_i64(cv: &CommunicationValue, dt: DataType) -> Option<i64> {
cv.get_data(dt)
.as_number()
.and_then(|n| i64::try_from(n).ok())
.or_else(|| {
cv.get_data(dt)
.as_signed_number()
.and_then(|n| i64::try_from(n).ok())
})
.or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::<i64>().ok()))
}
#[derive(Debug, Clone)]
struct ChatSecretRecipient {
user_id: String,
encrypted_secret: Vec<u8>,
kem_ciphertext: Vec<u8>,
}
fn recipient_from_value(value: &DataValue) -> Option<ChatSecretRecipient> {
let tm = TypeMap::latest();
let user_id = value
.get_field(DataType::UserId.to_id(&tm))?
.as_str()
.map(|s| s.to_string())
.or_else(|| {
value
.get_field(DataType::UserId.to_id(&tm))?
.as_number()
.map(|n| n.to_string())
})?;
let encrypted_secret = value
.get_field(DataType::EncryptedSecret.to_id(&tm))?
.as_bytes()?;
let kem_ciphertext = value
.get_field(DataType::KemCiphertext.to_id(&tm))?
.as_bytes()?;
Some(ChatSecretRecipient {
user_id,
encrypted_secret,
kem_ciphertext,
})
}
fn chat_secret_recipients(cv: &CommunicationValue) -> Option<Vec<ChatSecretRecipient>> {
let recipients = cv.get_data(DataType::Recipients).as_array()?;
let parsed = recipients
.iter()
.map(recipient_from_value)
.collect::<Option<Vec<_>>>()?;
if parsed.is_empty() {
None
} else {
Some(parsed)
}
}
fn set_chat_secret_cv_for_recipient(
source: &CommunicationValue,
recipient: &ChatSecretRecipient,
) -> CommunicationValue {
let recipient_value = typed_container(vec![
(DataType::UserId, DataValue::Str(recipient.user_id.clone())),
(
DataType::EncryptedSecret,
DataValue::Bytes(recipient.encrypted_secret.clone()),
),
(
DataType::KemCiphertext,
DataValue::Bytes(recipient.kem_ciphertext.clone()),
),
]);
CommunicationValue::new(CommunicationType::SetChatSecret)
.with_id(source.get_id())
.with_sender(source.get_sender())
.with_receiver(recipient.user_id.parse::<u64>().unwrap_or(0))
.add_typed_default(DataType::ChatId, source.get_data(DataType::ChatId).clone())
.add_typed_default(
DataType::SecretId,
source.get_data(DataType::SecretId).clone(),
)
.add_typed_default(
DataType::VersionNumber,
source.get_data(DataType::VersionNumber).clone(),
)
.add_typed_default(
DataType::WrappingScheme,
source.get_data(DataType::WrappingScheme).clone(),
)
.add_typed_default(
DataType::CreatedAt,
source.get_data(DataType::CreatedAt).clone(),
)
.add_typed_default(
DataType::Recipients,
DataValue::Array(vec![recipient_value]),
)
}
fn now_millis_i64() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as i64
}
fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue {
CommunicationValue::new(ty)
.with_id(request.get_id())
.with_receiver(request.get_sender())
}
// ============================================================================
// Waiting Task System
// ============================================================================
@ -85,7 +223,7 @@ impl ClientConnection {
}
if let Some(sender) = self.sender.read().await.as_ref() {
sender.close().await;
sender.close();
}
*self.sender.write().await = None;
@ -97,9 +235,12 @@ impl ClientConnection {
async fn handle_ping(self: Arc<Self>, cv: CommunicationValue) {
// Update our ping if provided
if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) {
let current = now_millis_i64();
let current = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis();
let mut ping_guard = self.ping.write().await;
*ping_guard = current - *last_ping as i64;
*ping_guard = current as i64 - *last_ping as i64;
}
// Send pong response
@ -183,8 +324,70 @@ impl ClientConnection {
}
if cv.is_type(CommunicationType::GetChatSecret) {
self.send_message(&message_handlers::handle_get_chat_secret(&cv))
.await;
let Some(user_id) = data_string(&cv, DataType::UserId) else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
};
let sender_id = cv.get_sender().to_string();
if user_id != sender_id {
self.send_message(&error_response(&cv, CommunicationType::ErrorNotFound))
.await;
return;
}
let Some(chat_id) = data_string(&cv, DataType::ChatId) else {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
return;
};
match e2ee_storage::get_chat_secret(ChatSecretQuery {
user_id,
chat_id,
secret_id: data_string(&cv, DataType::SecretId),
}) {
Ok(Some(record)) => {
let response = CommunicationValue::new(CommunicationType::ChatSecretResponse)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::UserId, DataValue::Str(record.user_id))
.add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id))
.add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id))
.add_typed_default(
DataType::VersionNumber,
DataValue::SignedNumber(record.version as i128),
)
.add_typed_default(
DataType::EncryptedSecret,
DataValue::Bytes(record.encrypted_secret),
)
.add_typed_default(
DataType::KemCiphertext,
DataValue::Bytes(record.kem_ciphertext),
)
.add_typed_default(
DataType::WrappingScheme,
DataValue::Str(record.wrapping_scheme),
)
.add_typed_default(
DataType::CreatedAt,
DataValue::SignedNumber(record.created_at as i128),
)
.add_typed_default(
DataType::UpdatedAt,
DataValue::SignedNumber(record.updated_at as i128),
);
self.send_message(&response).await;
}
Ok(None) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorNotSet))
.await;
}
Err(_) => {
self.send_message(&error_response(&cv, CommunicationType::ErrorInvalidData))
.await;
}
}
return;
}
@ -231,20 +434,132 @@ impl ClientConnection {
}
if cv.is_type(CommunicationType::CreateApp) {
self.send_message(&message_handlers::handle_create_app(&cv))
.await;
let sender_id = cv.get_sender() as i64;
let app_identifier = cv
.get_data(DataType::AppIdentifier)
.as_str()
.unwrap_or("")
.to_string();
let app_public_key = cv
.get_data(DataType::AppPublicKey)
.as_str()
.unwrap_or("")
.to_string();
if !app_identifier.is_empty() && !app_public_key.is_empty() {
if let Some(mut user) = iota_storage::users::user_manager::get_user(sender_id) {
if !user.trusted_apps.contains_key(&app_identifier) {
user.trusted_apps.insert(app_identifier, app_public_key);
iota_storage::users::user_manager::update_user(user);
}
}
}
let res = CommunicationValue::new(CommunicationType::CreateApp)
.with_id(cv.get_id())
.with_receiver(sender_id as u64);
self.send_message(&res).await;
return;
}
if cv.is_type(CommunicationType::DeleteApp) {
self.send_message(&message_handlers::handle_delete_app(&cv))
.await;
let sender_id = cv.get_sender() as i64;
let app_identifier = cv
.get_data(DataType::AppIdentifier)
.as_str()
.unwrap_or("")
.to_string();
if !app_identifier.is_empty() {
if let Some(mut user) = iota_storage::users::user_manager::get_user(sender_id) {
if user.trusted_apps.contains_key(&app_identifier) {
user.trusted_apps.remove(&app_identifier);
iota_storage::users::user_manager::update_user(user);
}
}
}
let res = CommunicationValue::new(CommunicationType::DeleteApp)
.with_id(cv.get_id())
.with_receiver(sender_id as u64);
self.send_message(&res).await;
return;
}
if cv.is_type(CommunicationType::ClientConnected) {
self.send_message(&message_handlers::handle_client_connected(&cv))
.await;
let user_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0) as i64;
let _session_id = cv.get_data(DataType::SessionId).as_number().unwrap_or(0) as i64;
let contacts = chats_util::get_users(user_id);
let mut contacts_array = Vec::new();
for (i, contact) in contacts.iter().enumerate() {
let mut contact_container = Vec::new();
contact_container.push((
DataType::UserId,
DataValue::SignedNumber(contact.user_id as i128),
));
contact_container.push((
DataType::LastMessageAt,
DataValue::SignedNumber(contact.last_message_at.unwrap_or(0) as i128),
));
if let Some(ref name) = contact.user_name {
contact_container.push((DataType::Username, DataValue::Str(name.clone())));
}
let amount = if i < 10 { 20 } else { 1 };
let messages = chat_files::get_messages(user_id, contact.user_id, 0, amount);
let mut msg_array = Vec::new();
for m in messages.members() {
let message_time = m["message_time"].as_i64().unwrap_or(0);
let content = m["content"].as_str().unwrap_or("").to_string();
let sent_by_self = m["sent_by_self"].as_bool().unwrap_or(false);
let height = m["height"].as_i64().unwrap_or(0);
let message_state = m["message_state"].as_str().unwrap_or("").to_string();
let mut msg_container = Vec::new();
msg_container.push((
DataType::SendTime,
DataValue::SignedNumber(message_time as i128),
));
msg_container.push((DataType::Content, DataValue::Str(content.clone())));
msg_container.push((DataType::MessageState, DataValue::Str(message_state)));
msg_container.push((DataType::Height, DataValue::SignedNumber(height as i128)));
msg_container.push((
DataType::SenderId,
DataValue::UnsignedNumber(if sent_by_self {
user_id as u128
} else {
contact.user_id as u128
}),
));
msg_array.push(typed_container(msg_container));
if msg_array.len() == 1 {
let sender_id = if sent_by_self {
user_id
} else {
contact.user_id
};
let mut last_msg = Vec::new();
last_msg.push((DataType::Content, DataValue::Str(content)));
last_msg.push((
DataType::SenderId,
DataValue::SignedNumber(sender_id as i128),
));
contact_container.push((DataType::LastMessage, typed_container(last_msg)));
}
}
contact_container.push((DataType::Messages, DataValue::Array(msg_array)));
contacts_array.push(typed_container(contact_container));
}
let resp = CommunicationValue::new(CommunicationType::ClientConnected)
.with_id(cv.get_id())
.add_typed_default(DataType::Contacts, DataValue::Array(contacts_array));
self.send_message(&resp).await;
return;
}
@ -253,32 +568,32 @@ impl ClientConnection {
// ************************************************ //
if cv.is_type(CommunicationType::MessageState) {
message_handlers::handle_message_state(&cv);
return;
}
let sender_id = &cv.get_sender();
let receiver_id = match cv.get_data(DataType::ChatPartnerId).as_number() {
Some(id) => id,
_ => return,
};
if cv.is_type(CommunicationType::MessageEdit) {
self.send_message(&message_handlers::handle_message_edit(&cv))
.await;
return;
}
// Parse send_time robustly: accept numeric or string, fallback to current time
let send_time_val = cv.get_data(DataType::SendTime);
let now_i64 = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as i64;
let timestamp_i64 = if let Some(n) = send_time_val.as_number() {
n as i64
} else if let Some(s) = send_time_val.as_str() {
s.parse::<i64>().unwrap_or(now_i64)
} else {
now_i64
};
if cv.is_type(CommunicationType::MessageReactionAdd) {
self.send_message(&message_handlers::handle_message_reaction(&cv, true))
.await;
return;
}
if cv.is_type(CommunicationType::MessageReactionRemove) {
self.send_message(&message_handlers::handle_message_reaction(&cv, false))
.await;
return;
}
if cv.is_type(CommunicationType::MessageDeleteLive) {
self.send_message(&message_handlers::handle_message_delete(&cv))
.await;
return;
let _ = chat_files::change_message_state(
timestamp_i64,
receiver_id as i64,
*sender_id as i64,
MessageState::from_str(cv.get_data(DataType::MessageState).as_str().unwrap_or("")),
);
}
// Incoming storsed message: store for the recipient, attempt local delivery, notify sender.
@ -294,7 +609,10 @@ impl ClientConnection {
// parse send_time safely (number or string), fallback to now
let send_time_val = cv.get_data(DataType::SendTime);
let now_i64 = now_millis_i64();
let now_i64 = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as i64;
let timestamp = if let Some(n) = send_time_val.as_number() {
n as i64
} else if let Some(s) = send_time_val.as_str() {
@ -420,38 +738,186 @@ impl ClientConnection {
}
if cv.is_type(CommunicationType::MessagesGet) {
self.send_message(&message_handlers::handle_messages_get(&cv))
.await;
let my_id = cv.get_sender();
let partner_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0);
let offset = cv.get_data(DataType::Offset).as_number().unwrap_or(0);
let amount = cv.get_data(DataType::Amount).as_number().unwrap_or(0);
let messages = chat_files::get_messages(
my_id as i64,
partner_id as i64,
offset as i64,
amount as i64,
);
let mut msg_array: Vec<DataValue> = Vec::new();
for m in messages.members() {
let message_time: i64 = m["message_time"].as_i64().unwrap_or(0);
let content: String = m["content"].as_str().unwrap_or("").to_string();
let sent_by_self: bool = m["sent_by_self"].as_bool().unwrap_or(false);
let height: i64 = m["height"].as_i64().unwrap_or(0);
let sender_id: i64 = if sent_by_self {
my_id as i64
} else {
if let Some(n) = cv.get_data(DataType::ChatPartnerId).as_number() {
n as i64
} else if let Some(s) = cv.get_data(DataType::ChatPartnerId).as_str() {
s.parse::<i64>().unwrap_or(partner_id as i64)
} else {
partner_id as i64
}
};
let message_state: String = m["message_state"].as_str().unwrap_or("").to_string();
let mut container = Vec::new();
container.push((
DataType::SendTime,
DataValue::SignedNumber(message_time as i128),
));
container.push((DataType::Content, DataValue::Str(content)));
container.push((
DataType::SenderId,
DataValue::SignedNumber(sender_id as i128),
));
container.push((DataType::MessageState, DataValue::Str(message_state)));
container.push((DataType::Height, DataValue::SignedNumber(height as i128)));
container.push((
DataType::SenderId,
DataValue::UnsignedNumber(if sent_by_self {
my_id as u128
} else {
partner_id as u128
}),
));
msg_array.push(typed_container(container));
}
let resp = CommunicationValue::new(CommunicationType::MessagesGet)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(DataType::Messages, DataValue::Array(msg_array));
self.send_message(&resp).await;
return;
}
if cv.is_type(CommunicationType::GetChats) {
self.send_message(&message_handlers::handle_get_chats(&cv))
.await;
let user_id = cv.get_sender();
let users = chats_util::get_users(user_id as i64);
let mut user_array = Vec::new();
for user in users {
let mut container = Vec::new();
container.push((
DataType::UserId,
DataValue::SignedNumber(user.user_id as i128),
));
if let Some(name) = user.user_name {
container.push((DataType::Username, DataValue::Str(name)));
}
if let Some(ts) = user.last_message_at {
container.push((DataType::LastMessageAt, DataValue::SignedNumber(ts as i128)));
}
user_array.push(typed_container(container));
}
let resp = CommunicationValue::new(CommunicationType::GetChats)
.with_id(cv.get_id())
.with_receiver(user_id)
.add_typed_default(DataType::UserIds, DataValue::Array(user_array));
self.send_message(&resp).await;
return;
}
if cv.is_type(CommunicationType::AddConversation) {
self.send_message(&message_handlers::handle_add_conversation(&cv))
.await;
let user_id = cv.get_sender();
let other_id = match cv.get_data(DataType::ChatPartnerId).as_number() {
Some(n) => n as i64,
None => cv
.get_data(DataType::ChatPartnerId)
.as_str()
.unwrap_or("0")
.parse()
.unwrap_or(0),
};
let mut contact = get_user(user_id as i64, other_id).unwrap_or(Contact::new(other_id));
if let Some(name) = cv.get_data(DataType::ChatPartnerName).as_str() {
contact.user_name = Some(name.to_string());
}
contact.set_last_message_at(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as i64,
);
mod_user(user_id as i64, &contact);
let resp = CommunicationValue::new(CommunicationType::AddConversation)
.with_id(cv.get_id())
.with_receiver(user_id);
self.send_message(&resp).await;
return;
}
if cv.is_type(CommunicationType::AddCommunity) {
self.send_message(&message_handlers::handle_add_community(&cv))
.await;
CommunitiesUtil::add_community(
cv.get_sender() as i64,
cv.get_data(DataType::CommunityAddress)
.as_str()
.unwrap()
.to_string(),
cv.get_data(DataType::CommunityTitle)
.as_str()
.unwrap()
.to_string(),
cv.get_data(DataType::Position)
.as_str()
.unwrap()
.to_string(),
);
let resp = CommunicationValue::new(CommunicationType::AddCommunity)
.with_id(cv.get_id())
.with_receiver(cv.get_sender());
self.send_message(&resp).await;
return;
}
if cv.is_type(CommunicationType::GetCommunities) {
self.send_message(&message_handlers::handle_get_communities(&cv))
.await;
let mut comm_array = Vec::new();
for c in CommunitiesUtil::get_communities(cv.get_sender() as i64) {
let mut container: Vec<(DataType, DataValue)> = Vec::new();
if let Some(address) = c["address"].as_str() {
container.push((
DataType::CommunityAddress,
DataValue::Str(address.to_string()),
));
}
if let Some(title) = c["title"].as_str() {
container.push((DataType::CommunityTitle, DataValue::Str(title.to_string())));
}
if let Some(position) = c["position"].as_str() {
container.push((DataType::Position, DataValue::Str(position.to_string())));
}
comm_array.push(typed_container(container));
}
let resp = CommunicationValue::new(CommunicationType::GetCommunities)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::Communities, DataValue::Array(comm_array));
self.send_message(&resp).await;
return;
}
if cv.is_type(CommunicationType::RemoveCommunity) {
self.send_message(&message_handlers::handle_remove_community(&cv))
.await;
CommunitiesUtil::remove_community(
cv.get_sender() as i64,
cv.get_data(DataType::CommunityAddress)
.as_str()
.unwrap()
.to_string(),
);
let resp = CommunicationValue::new(CommunicationType::RemoveCommunity)
.with_id(cv.get_id())
.with_receiver(cv.get_sender());
self.send_message(&resp).await;
return;
}
@ -460,11 +926,10 @@ impl ClientConnection {
let settings_name = cv.get_data(DataType::SettingsName).as_str().unwrap();
let settings_value = cv.get_data(DataType::Payload).as_str().unwrap();
let _ = iota_storage::util::settings::save(
my_id as i64,
iota_storage::util::settings::GLOBAL_SESSION_ID,
settings_name,
settings_value,
save_file(
&format!("users/{}/settings/", my_id),
&format!("{}.settings", settings_name),
&settings_value,
);
let response = CommunicationValue::new(CommunicationType::SettingsSave)
@ -478,14 +943,10 @@ impl ClientConnection {
if cv.is_type(CommunicationType::SettingsLoad) {
let my_id = cv.get_sender();
let settings_name = cv.get_data(DataType::SettingsName).as_string().unwrap();
let settings_value_str = iota_storage::util::settings::load(
my_id as i64,
iota_storage::util::settings::GLOBAL_SESSION_ID,
&settings_name,
)
.ok()
.flatten()
.unwrap_or_default();
let settings_value_str = load_file(
&format!("users/{}/settings/", my_id),
&format!("{}.settings", settings_name),
);
let response = CommunicationValue::new(CommunicationType::SettingsLoad)
.with_id(cv.get_id())
.with_receiver(my_id)
@ -498,12 +959,15 @@ impl ClientConnection {
if cv.is_type(CommunicationType::SettingsList) {
let my_id = cv.get_sender();
let settings = iota_storage::util::settings::list(
my_id as i64,
iota_storage::util::settings::GLOBAL_SESSION_ID,
)
.unwrap_or_default();
let settings_json = settings.into_iter().map(DataValue::Str).collect();
let settings = get_children(&format!("users/{}/settings/", my_id));
let mut settings_json = Vec::new();
for s in settings {
let s = s.replace(".settings", "");
if s.is_empty() {
continue;
}
let _ = settings_json.push(DataValue::Str(s));
}
let response = CommunicationValue::new(CommunicationType::SettingsList)
.with_id(cv.get_id())
.with_receiver(my_id)
@ -550,7 +1014,7 @@ impl ClientConnection {
if !sender.is_open() {
drop(sender_guard);
if let Some(sender) = self.sender.write().await.take() {
sender.close().await;
sender.close();
}
return Err("connection closed".to_string());
}

2
iota-cli/src/elements/console_card.rs
View file

@ -444,7 +444,7 @@ pub async fn run_command(command: &str) {
if let Some(user) = user_manager::get_user_by_username(username) {
let msg = CommunicationValue::new(CommunicationType::DeleteUser)
.with_sender(user.user_id as u64);
let _ = OMIKRON_CONNECTION.send_message(&msg).await;
OMIKRON_CONNECTION.send_message(&msg).await;
user_manager::remove_user(user.user_id);
log!("Removed user {}", user.user_id);
} else {

9
iota-connection/Cargo.toml
View file

@ -1,9 +0,0 @@
[package]
name = "iota-connection"
version = "0.1.0"
edition = "2024"
[dependencies]
iota-storage = { path = "../iota-storage" }
iota-util = { path = "../iota-util" }
mtp = { git = "https://git.methanium.net/Methanium/mtp.git" }

36
iota-connection/src/connection_handler.rs
View file

@ -1,36 +0,0 @@
use mtp::codec::CommunicationValue;
use std::future::Future;
use std::time::Duration;
/// Unified interface for all connection types (Omikron, Direct, future modes).
///
/// Provides the common messaging API that the rest of the codebase uses,
/// regardless of whether the connection goes through Omikron or is direct.
pub trait ConnectionHandler: Send + Sync {
/// Send a message to the remote end.
fn send_message(
&self,
cv: &CommunicationValue,
) -> impl Future<Output = Result<(), String>> + Send;
/// Send a message and wait for a correlated response.
///
/// The implementation correlates requests/responses by message ID and
/// enforces the given `timeout`. Returns an error on timeout or if the
/// connection drops while waiting.
fn await_response(
&self,
cv: &CommunicationValue,
timeout: Option<Duration>,
) -> impl Future<Output = Result<CommunicationValue, String>> + Send;
/// Returns `true` when the connection is alive and ready for traffic.
fn is_connected(&self) -> impl Future<Output = bool> + Send;
/// Returns `true` when the connection has completed identification /
/// registration and is fully operational.
fn is_identified(&self) -> impl Future<Output = bool> + Send;
/// Gracefully tear down the connection.
fn stop(&self) -> impl Future<Output = ()> + Send;
}

3
iota-connection/src/lib.rs
View file

@ -1,3 +0,0 @@
pub mod connection_handler;
pub mod message_common;
pub mod message_handlers;

137
iota-connection/src/message_common.rs
View file

@ -1,137 +0,0 @@
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use mtp::type_map::TypeMap;
use std::time::{SystemTime, UNIX_EPOCH};
pub fn typed_container(items: Vec<(DataType, DataValue)>) -> DataValue {
use mtp::type_map::{DataTypeId, TypeMap};
let tm = TypeMap::latest();
DataValue::Container(
items
.into_iter()
.filter_map(|(dt, dv)| tm.data_id_enum(dt).map(|id| (DataTypeId(id), dv)))
.collect(),
)
}
pub fn data_string(cv: &CommunicationValue, dt: DataType) -> Option<String> {
cv.get_data(dt)
.as_str()
.map(|s| s.to_string())
.or_else(|| cv.get_data(dt).as_number().map(|n| n.to_string()))
.or_else(|| cv.get_data(dt).as_signed_number().map(|n| n.to_string()))
}
pub fn data_i64(cv: &CommunicationValue, dt: DataType) -> Option<i64> {
cv.get_data(dt)
.as_number()
.and_then(|n| i64::try_from(n).ok())
.or_else(|| {
cv.get_data(dt)
.as_signed_number()
.and_then(|n| i64::try_from(n).ok())
})
.or_else(|| cv.get_data(dt).as_str().and_then(|s| s.parse::<i64>().ok()))
}
#[derive(Debug, Clone)]
pub struct ChatSecretRecipient {
pub user_id: String,
pub encrypted_secret: Vec<u8>,
pub kem_ciphertext: Vec<u8>,
}
pub fn recipient_from_value(value: &DataValue) -> Option<ChatSecretRecipient> {
let tm = TypeMap::latest();
let user_id = value
.get_field(DataType::UserId.try_to_id(&tm)?)?
.as_str()
.map(|s| s.to_string())
.or_else(|| {
value
.get_field(DataType::UserId.try_to_id(&tm)?)?
.as_number()
.map(|n| n.to_string())
})?;
let encrypted_secret = value
.get_field(DataType::EncryptedSecret.try_to_id(&tm)?)?
.as_bytes()?;
let kem_ciphertext = value
.get_field(DataType::KemCiphertext.try_to_id(&tm)?)?
.as_bytes()?;
Some(ChatSecretRecipient {
user_id,
encrypted_secret,
kem_ciphertext,
})
}
pub fn chat_secret_recipients(cv: &CommunicationValue) -> Option<Vec<ChatSecretRecipient>> {
let recipients = cv.get_data(DataType::Recipients).as_array()?;
let parsed = recipients
.iter()
.map(recipient_from_value)
.collect::<Option<Vec<_>>>()?;
if parsed.is_empty() {
None
} else {
Some(parsed)
}
}
pub fn set_chat_secret_cv_for_recipient(
source: &CommunicationValue,
recipient: &ChatSecretRecipient,
) -> CommunicationValue {
let recipient_value = typed_container(vec![
(DataType::UserId, DataValue::Str(recipient.user_id.clone())),
(
DataType::EncryptedSecret,
DataValue::Bytes(recipient.encrypted_secret.clone()),
),
(
DataType::KemCiphertext,
DataValue::Bytes(recipient.kem_ciphertext.clone()),
),
]);
CommunicationValue::new(CommunicationType::SetChatSecret)
.with_id(source.get_id())
.with_sender(source.get_sender())
.with_receiver(recipient.user_id.parse::<u64>().unwrap_or(0))
.add_typed_default(DataType::ChatId, source.get_data(DataType::ChatId).clone())
.add_typed_default(
DataType::SecretId,
source.get_data(DataType::SecretId).clone(),
)
.add_typed_default(
DataType::VersionNumber,
source.get_data(DataType::VersionNumber).clone(),
)
.add_typed_default(
DataType::WrappingScheme,
source.get_data(DataType::WrappingScheme).clone(),
)
.add_typed_default(
DataType::CreatedAt,
source.get_data(DataType::CreatedAt).clone(),
)
.add_typed_default(
DataType::Recipients,
DataValue::Array(vec![recipient_value]),
)
}
pub fn now_millis_i64() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as i64
}
pub fn error_response(request: &CommunicationValue, ty: CommunicationType) -> CommunicationValue {
CommunicationValue::new(ty)
.with_id(request.get_id())
.with_receiver(request.get_sender())
}

766
iota-connection/src/message_handlers.rs
View file

@ -1,766 +0,0 @@
use crate::message_common::*;
use iota_storage::util::chat_files::{self, MessageState};
use iota_storage::util::chats_util::{self, get_user, mod_user};
use iota_storage::util::communities_util::CommunitiesUtil;
use iota_storage::util::e2ee_storage::{self, ChatSecretQuery};
use iota_storage::util::settings;
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
pub struct MessageMutation {
pub sender_id: i64,
pub partner_id: i64,
pub send_time: i64,
}
pub fn message_mutation(cv: &CommunicationValue) -> Result<MessageMutation, CommunicationValue> {
let sender_id = i64::try_from(cv.get_sender())
.map_err(|_| error_response(cv, CommunicationType::ErrorInvalidData))?;
let partner_id = data_i64(cv, DataType::ChatPartnerId)
.filter(|id| *id > 0)
.ok_or_else(|| error_response(cv, CommunicationType::ErrorInvalidData))?;
let send_time = data_i64(cv, DataType::SendTime)
.filter(|time| *time > 0)
.ok_or_else(|| error_response(cv, CommunicationType::ErrorInvalidData))?;
Ok(MessageMutation {
sender_id,
partner_id,
send_time,
})
}
pub fn success_response(cv: &CommunicationValue) -> CommunicationValue {
error_response(cv, CommunicationType::Success)
}
pub fn handle_message_edit(cv: &CommunicationValue) -> CommunicationValue {
let mutation = match message_mutation(cv) {
Ok(mutation) => mutation,
Err(response) => return response,
};
let Some(content) = cv.get_data(DataType::Content).as_str() else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
match chat_files::edit_message(
mutation.sender_id,
mutation.partner_id,
mutation.send_time,
mutation.sender_id,
content,
) {
Ok(()) => success_response(cv),
Err(_) => error_response(cv, CommunicationType::ErrorNotFound),
}
}
pub fn handle_message_reaction(cv: &CommunicationValue, add: bool) -> CommunicationValue {
let mutation = match message_mutation(cv) {
Ok(mutation) => mutation,
Err(response) => return response,
};
let Some(reaction) = cv.get_data(DataType::Reaction).as_str() else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
if reaction.is_empty() || reaction.len() > 64 {
return error_response(cv, CommunicationType::ErrorInvalidData);
}
let result = if add {
chat_files::add_reaction(
mutation.sender_id,
mutation.partner_id,
mutation.send_time,
mutation.sender_id,
reaction,
)
} else {
chat_files::remove_reaction(
mutation.sender_id,
mutation.partner_id,
mutation.send_time,
mutation.sender_id,
reaction,
)
};
match result {
Ok(()) => success_response(cv),
Err(_) => error_response(cv, CommunicationType::ErrorNotFound),
}
}
pub fn handle_message_delete(cv: &CommunicationValue) -> CommunicationValue {
let mutation = match message_mutation(cv) {
Ok(mutation) => mutation,
Err(response) => return response,
};
match chat_files::delete_message(mutation.sender_id, mutation.partner_id, mutation.send_time) {
Ok(()) => success_response(cv),
Err(_) => error_response(cv, CommunicationType::ErrorNotFound),
}
}
fn stored_message_value(
message: &chat_files::StoredMessage,
storage_owner: i64,
partner_id: i64,
) -> DataValue {
let mut fields = vec![
(
DataType::SendTime,
DataValue::SignedNumber(message.message_time as i128),
),
(DataType::Content, DataValue::Str(message.content.clone())),
(
DataType::MessageState,
DataValue::Str(message.message_state.clone()),
),
(
DataType::Height,
DataValue::SignedNumber(message.height as i128),
),
(
DataType::SenderId,
DataValue::UnsignedNumber(if message.sent_by_self {
storage_owner as u128
} else {
partner_id as u128
}),
),
];
if message.edited {
fields.push((DataType::Edited, DataValue::Bool(true)));
}
if let Some(reply_to) = message.reply_to {
fields.push((
DataType::ReplyId,
DataValue::UnsignedNumber(reply_to as u64 as u128),
));
}
if !message.reactions.is_empty() {
let reactions = message
.reactions
.iter()
.map(|reaction| {
typed_container(vec![
(
DataType::Reaction,
DataValue::Str(reaction.reaction.clone()),
),
(
DataType::SenderId,
DataValue::SignedNumber(reaction.user_id as i128),
),
])
})
.collect();
fields.push((DataType::Reactions, DataValue::Array(reactions)));
}
typed_container(fields)
}
pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue {
let Some(user_id) = data_string(cv, DataType::UserId) else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
if user_id != cv.get_sender().to_string() {
return error_response(cv, CommunicationType::ErrorNotFound);
}
let Some(chat_id) = data_string(cv, DataType::ChatId) else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
match e2ee_storage::get_chat_secret(ChatSecretQuery {
user_id,
chat_id,
secret_id: data_string(cv, DataType::SecretId),
}) {
Ok(Some(record)) => CommunicationValue::new(CommunicationType::ChatSecretResponse)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::UserId, DataValue::Str(record.user_id))
.add_typed_default(DataType::ChatId, DataValue::Str(record.chat_id))
.add_typed_default(DataType::SecretId, DataValue::Str(record.secret_id))
.add_typed_default(
DataType::VersionNumber,
DataValue::SignedNumber(record.version as i128),
)
.add_typed_default(
DataType::EncryptedSecret,
DataValue::Bytes(record.encrypted_secret),
)
.add_typed_default(
DataType::KemCiphertext,
DataValue::Bytes(record.kem_ciphertext),
)
.add_typed_default(
DataType::WrappingScheme,
DataValue::Str(record.wrapping_scheme),
)
.add_typed_default(
DataType::CreatedAt,
DataValue::SignedNumber(record.created_at as i128),
)
.add_typed_default(
DataType::UpdatedAt,
DataValue::SignedNumber(record.updated_at as i128),
),
Ok(None) => error_response(cv, CommunicationType::ErrorNotSet),
Err(_) => error_response(cv, CommunicationType::ErrorInvalidData),
}
}
pub fn handle_create_app(cv: &CommunicationValue) -> CommunicationValue {
let sender_id = cv.get_sender() as i64;
let app_identifier = cv
.get_data(DataType::AppIdentifier)
.as_str()
.unwrap_or("")
.to_string();
let app_public_key = cv
.get_data(DataType::AppPublicKey)
.as_str()
.unwrap_or("")
.to_string();
if !app_identifier.is_empty() && !app_public_key.is_empty() {
if let Some(mut user) = iota_storage::users::user_manager::get_user(sender_id) {
if !user.trusted_apps.contains_key(&app_identifier) {
user.trusted_apps.insert(app_identifier, app_public_key);
iota_storage::users::user_manager::update_user(user);
}
}
}
CommunicationValue::new(CommunicationType::CreateApp)
.with_id(cv.get_id())
.with_receiver(sender_id as u64)
}
pub fn handle_delete_app(cv: &CommunicationValue) -> CommunicationValue {
let sender_id = cv.get_sender() as i64;
let app_identifier = cv
.get_data(DataType::AppIdentifier)
.as_str()
.unwrap_or("")
.to_string();
if !app_identifier.is_empty() {
if let Some(mut user) = iota_storage::users::user_manager::get_user(sender_id) {
if user.trusted_apps.contains_key(&app_identifier) {
user.trusted_apps.remove(&app_identifier);
iota_storage::users::user_manager::update_user(user);
}
}
}
CommunicationValue::new(CommunicationType::DeleteApp)
.with_id(cv.get_id())
.with_receiver(sender_id as u64)
}
pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
let user_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0) as i64;
let contacts = chats_util::get_users(user_id);
let mut contacts_array = Vec::new();
for (i, contact) in contacts.iter().enumerate() {
let mut contact_container = Vec::new();
contact_container.push((
DataType::UserId,
DataValue::SignedNumber(contact.user_id as i128),
));
contact_container.push((
DataType::LastMessageAt,
DataValue::SignedNumber(contact.last_message_at.unwrap_or(0) as i128),
));
if let Some(ref name) = contact.user_name {
contact_container.push((DataType::Username, DataValue::Str(name.clone())));
}
let amount = if i < 10 { 20 } else { 1 };
let messages = chat_files::get_messages(user_id, contact.user_id, 0, amount);
let mut msg_array = Vec::new();
for m in &messages {
msg_array.push(stored_message_value(m, user_id, contact.user_id));
if msg_array.len() == 1 {
let sender_id = if m.sent_by_self {
user_id
} else {
contact.user_id
};
let mut last_msg = Vec::new();
last_msg.push((DataType::Content, DataValue::Str(m.content.clone())));
last_msg.push((
DataType::SenderId,
DataValue::SignedNumber(sender_id as i128),
));
contact_container.push((DataType::LastMessage, typed_container(last_msg)));
}
}
contact_container.push((DataType::Messages, DataValue::Array(msg_array)));
contacts_array.push(typed_container(contact_container));
}
CommunicationValue::new(CommunicationType::ClientConnected)
.with_id(cv.get_id())
.add_typed_default(DataType::Contacts, DataValue::Array(contacts_array))
}
pub fn handle_message_state(cv: &CommunicationValue) {
let sender_id = &cv.get_sender();
let receiver_id = match cv.get_data(DataType::ChatPartnerId).as_number() {
Some(id) => id,
_ => return,
};
let timestamp_i64 = if let Some(n) = cv.get_data(DataType::SendTime).as_number() {
n as i64
} else if let Some(s) = cv.get_data(DataType::SendTime).as_str() {
s.parse::<i64>().unwrap_or_else(|_| now_millis_i64())
} else {
now_millis_i64()
};
let _ = chat_files::change_message_state(
timestamp_i64,
receiver_id as i64,
*sender_id as i64,
MessageState::from_str(cv.get_data(DataType::MessageState).as_str().unwrap_or("")),
);
}
pub fn handle_messages_get(cv: &CommunicationValue) -> CommunicationValue {
let my_id = cv.get_sender();
let partner_id = cv.get_data(DataType::UserId).as_number().unwrap_or(0);
let offset = cv.get_data(DataType::Offset).as_number().unwrap_or(0);
let amount = cv.get_data(DataType::Amount).as_number().unwrap_or(0);
let messages = chat_files::get_messages(
my_id as i64,
partner_id as i64,
offset as i64,
amount as i64,
);
let mut msg_array: Vec<DataValue> = Vec::new();
for m in &messages {
msg_array.push(stored_message_value(m, my_id as i64, partner_id as i64));
}
CommunicationValue::new(CommunicationType::MessagesGet)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(DataType::Messages, DataValue::Array(msg_array))
}
pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue {
let user_id = cv.get_sender();
let users = chats_util::get_users(user_id as i64);
let mut user_array = Vec::new();
for user in users {
let mut container = Vec::new();
container.push((
DataType::UserId,
DataValue::SignedNumber(user.user_id as i128),
));
if let Some(name) = user.user_name {
container.push((DataType::Username, DataValue::Str(name)));
}
if let Some(ts) = user.last_message_at {
container.push((DataType::LastMessageAt, DataValue::SignedNumber(ts as i128)));
}
user_array.push(typed_container(container));
}
CommunicationValue::new(CommunicationType::GetChats)
.with_id(cv.get_id())
.with_receiver(user_id)
.add_typed_default(DataType::UserIds, DataValue::Array(user_array))
}
pub fn handle_add_conversation(cv: &CommunicationValue) -> CommunicationValue {
let user_id = cv.get_sender();
let other_id = match cv.get_data(DataType::ChatPartnerId).as_number() {
Some(n) => n as i64,
None => cv
.get_data(DataType::ChatPartnerId)
.as_str()
.unwrap_or("0")
.parse()
.unwrap_or(0),
};
let mut contact = get_user(user_id as i64, other_id)
.unwrap_or(iota_storage::users::contact::Contact::new(other_id));
if let Some(name) = cv.get_data(DataType::ChatPartnerName).as_str() {
contact.user_name = Some(name.to_string());
}
contact.set_last_message_at(now_millis_i64());
mod_user(user_id as i64, &contact);
CommunicationValue::new(CommunicationType::AddConversation)
.with_id(cv.get_id())
.with_receiver(user_id)
}
pub fn handle_add_community(cv: &CommunicationValue) -> CommunicationValue {
CommunitiesUtil::add_community(
cv.get_sender() as i64,
cv.get_data(DataType::CommunityAddress)
.as_str()
.unwrap()
.to_string(),
cv.get_data(DataType::CommunityTitle)
.as_str()
.unwrap()
.to_string(),
cv.get_data(DataType::Position)
.as_str()
.unwrap()
.to_string(),
);
CommunicationValue::new(CommunicationType::AddCommunity)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
}
pub fn handle_get_communities(cv: &CommunicationValue) -> CommunicationValue {
let mut comm_array = Vec::new();
for c in CommunitiesUtil::get_communities(cv.get_sender() as i64) {
let mut container: Vec<(DataType, DataValue)> = Vec::new();
container.push((
DataType::CommunityAddress,
DataValue::Str(c.address.clone()),
));
container.push((DataType::CommunityTitle, DataValue::Str(c.title.clone())));
container.push((DataType::Position, DataValue::Str(c.position.clone())));
comm_array.push(typed_container(container));
}
CommunicationValue::new(CommunicationType::GetCommunities)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
.add_typed_default(DataType::Communities, DataValue::Array(comm_array))
}
pub fn handle_remove_community(cv: &CommunicationValue) -> CommunicationValue {
CommunitiesUtil::remove_community(
cv.get_sender() as i64,
cv.get_data(DataType::CommunityAddress)
.as_str()
.unwrap()
.to_string(),
);
CommunicationValue::new(CommunicationType::RemoveCommunity)
.with_id(cv.get_id())
.with_receiver(cv.get_sender())
}
pub fn handle_global_settings_save(cv: &CommunicationValue) -> CommunicationValue {
let my_id = cv.get_sender();
let Some(settings_value) = cv.get_data(DataType::Payload).as_str() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing settings payload".to_string()),
);
};
if settings::save_global(my_id as i64, settings_value).is_err() {
return error_response(cv, CommunicationType::ErrorInvalidData);
}
let mut response = CommunicationValue::new(CommunicationType::GlobalSettingsSave)
.with_receiver(my_id)
.with_id(cv.get_id());
if let Some(session_id) = cv.get_data(DataType::SessionId).as_number() {
response = response.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
response
}
pub fn handle_global_settings_load(cv: &CommunicationValue) -> CommunicationValue {
let my_id = cv.get_sender();
let Ok(settings_value) = settings::load_global(my_id as i64) else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
let Some(settings_value_str) = settings_value else {
let mut response = CommunicationValue::new(CommunicationType::ErrorNotFound)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Path,
DataValue::Str("global.settings".to_string()),
);
if let Some(session_id) = cv.get_data(DataType::SessionId).as_number() {
response = response.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
return response;
};
let mut response = CommunicationValue::new(CommunicationType::GlobalSettingsLoad)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(DataType::Payload, DataValue::Str(settings_value_str));
if let Some(session_id) = cv.get_data(DataType::SessionId).as_number() {
response = response.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
response
}
pub fn handle_settings_save(
cv: &CommunicationValue,
_expected_session_id: i128,
) -> CommunicationValue {
let my_id = cv.get_sender();
let Some(session_id) = cv.get_data(DataType::SessionId).as_number() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing session_id".to_string()),
);
};
if session_id == 0 || session_id > 1_000_000 {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Invalid session_id".to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
};
let Some(settings_name) = cv.get_data(DataType::SettingsName).as_str() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing settings_name".to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
};
let Some(settings_value) = cv.get_data(DataType::Payload).as_str() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing settings payload".to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
};
if !settings_name
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '-' || c == '.')
|| settings_name.contains("..")
{
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Invalid settings_name".to_string()),
)
.add_typed_default(
DataType::SettingsName,
DataValue::Str(settings_name.to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
if settings::save(
my_id as i64,
session_id as i64,
settings_name,
settings_value,
)
.is_err()
{
return error_response(cv, CommunicationType::ErrorInvalidData);
}
CommunicationValue::new(CommunicationType::SettingsSave)
.with_receiver(my_id)
.with_id(cv.get_id())
.add_typed_default(
DataType::SettingsName,
DataValue::Str(settings_name.to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
)
}
pub fn handle_settings_load(
cv: &CommunicationValue,
_expected_session_id: i128,
) -> CommunicationValue {
let my_id = cv.get_sender();
let Some(session_id) = cv.get_data(DataType::SessionId).as_number() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing session_id".to_string()),
);
};
if session_id == 0 || session_id > 1_000_000 {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Invalid session_id".to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
let Some(settings_name) = cv.get_data(DataType::SettingsName).as_str() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing settings_name".to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
};
if !settings_name
.chars()
.all(|c| c.is_alphanumeric() || c == '_' || c == '-' || c == '.')
|| settings_name.contains("..")
{
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Invalid settings_name".to_string()),
)
.add_typed_default(
DataType::SettingsName,
DataValue::Str(settings_name.to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
let Ok(settings_value) = settings::load(my_id as i64, session_id as i64, settings_name) else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
let Some(settings_value_str) = settings_value else {
return CommunicationValue::new(CommunicationType::ErrorNotFound)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::SettingsName,
DataValue::Str(settings_name.to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
};
CommunicationValue::new(CommunicationType::SettingsLoad)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(DataType::Payload, DataValue::Str(settings_value_str))
.add_typed_default(
DataType::SettingsName,
DataValue::Str(settings_name.to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
)
}
pub fn handle_settings_list(
cv: &CommunicationValue,
_expected_session_id: i128,
) -> CommunicationValue {
let my_id = cv.get_sender();
let Some(session_id) = cv.get_data(DataType::SessionId).as_number() else {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Missing session_id".to_string()),
);
};
if session_id == 0 || session_id > 1_000_000 {
return CommunicationValue::new(CommunicationType::ErrorInvalidData)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(
DataType::Message,
DataValue::Str("Invalid session_id".to_string()),
)
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
);
}
let Ok(settings) = settings::list(my_id as i64, session_id as i64) else {
return error_response(cv, CommunicationType::ErrorInvalidData);
};
let settings_json = settings.into_iter().map(DataValue::Str).collect();
CommunicationValue::new(CommunicationType::SettingsList)
.with_id(cv.get_id())
.with_receiver(my_id)
.add_typed_default(DataType::Settings, DataValue::Array(settings_json))
.add_typed_default(
DataType::SessionId,
DataValue::SignedNumber(session_id as i128),
)
}

29
iota-core/src/main.rs
View file

@ -88,9 +88,6 @@ async fn main() {
if let Err(_) = user_manager::load_users().await {
log_t!("user_load_failed");
}
if let Err(e) = iota_storage::util::settings::migrate_legacy_files() {
log!("Failed to migrate legacy settings: {}", e);
}
let mut sb = "".to_string();
@ -105,11 +102,7 @@ async fn main() {
}
log!(
"IOTA ID: {}",
CONFIG
.load()
.iota_id
.map(|id| id.to_string())
.unwrap_or_else(|| "N/A".to_string())
CONFIG.load().iota_id.map(|id| id.to_string()).unwrap_or_else(|| "N/A".to_string())
);
log!("User IDS: {}", sb);
@ -128,7 +121,7 @@ async fn main() {
sb1 = sb1 + ",";
}
log!("Community IDS: {}", sb1); */
let port = CONFIG.load().port;
let _port = CONFIG.load().port;
let mut _ip = "0.0.0.0".to_string();
for iface in pnet::datalink::interfaces() {
let iface: NetworkInterface = iface;
@ -140,6 +133,17 @@ async fn main() {
}
}
}
/* Community port activation is used for activating the port for communities.
* Code is currently commented because communities have not been implemented yet.
if start(port).await {
log_t!("community_active", ip, port.to_string());
} else {
if port < 1024 {
log_t!("community_start_error_admin", port.to_string());
} else {
log_t!("community_start_error", port.to_string());
}
} */
if !has_dir("web") {
download_and_extract_zip(
"https://omega.tensamin.net/api/download/iota_frontend",
@ -147,9 +151,6 @@ async fn main() {
)
.await;
}
if !web_server::start(port).await {
log!("Failed to start the MTP web server on port {}", port);
}
let _ = omikron::omikron_connection::get_omikron_connection().await;
log_t!("setup_completed");
@ -161,9 +162,7 @@ async fn main() {
if OMIKRON_CONNECTION.has_auth_failure().await {
if let Some(reason) = OMIKRON_CONNECTION.get_auth_failure().await {
log!("Authentication failed: {}", reason);
log!(
"Use /reconnect to try again or /regenerate private-key to create a new key pair"
);
log!("Use /reconnect to try again or /regenerate private-key to create a new key pair");
OMIKRON_CONNECTION.clear_auth_failure().await;
}
}

3
iota-storage/Cargo.toml
View file

@ -17,11 +17,8 @@ hkdf = "0.12.4"
json = "*"
arc-swap = "1"
once_cell = "1.21.3"
r2d2 = "0.8"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
serde_yaml = "0.9"
thiserror = "2"
rand = "0.8"
rand_core = { version = "0.6", features = ["getrandom", "std"] }
ratatui = "0.30.0"

1
iota-storage/src/lib.rs
View file

@ -1,3 +1,2 @@
pub mod storage_error;
pub mod users;
pub mod util;

13
iota-storage/src/storage_error.rs
View file

@ -1,13 +0,0 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum StorageError {
#[error("Database error: {0}")]
Db(#[from] rusqlite::Error),
#[error("Connection pool error: {0}")]
Pool(String),
#[error("IO error: {0}")]
Io(#[from] std::io::Error),
#[error("{0}")]
Other(String),
}

26
iota-storage/src/users/contact.rs
View file

@ -1,3 +1,4 @@
use json::{self, JsonValue, number::Number};
use std::time::{SystemTime, UNIX_EPOCH};
#[derive(Debug, Clone)]
@ -32,4 +33,29 @@ impl Contact {
pub fn set_last_message_at(&mut self, p0: i64) {
self.last_message_at = Option::from(p0);
}
pub fn to_json(&self) -> JsonValue {
let mut obj = JsonValue::new_object();
obj["user_id"] = JsonValue::Number(Number::from(self.user_id));
if let Some(name) = &self.user_name {
obj["user_name"] = JsonValue::from(name.as_str());
}
if let Some(ts) = &self.last_message_at {
obj["last_message_at"] = JsonValue::Number(Number::from(*ts));
}
obj
}
pub fn from_json(o: &JsonValue) -> Contact {
let user_id = o["user_id"].as_i64().unwrap_or(0);
let user_name = o["user_name"].as_str().map(|s| s.to_string());
let last_message_at = o["last_message_at"].as_i64();
Contact {
user_id,
user_name,
last_message_at,
}
}
}

1
iota-storage/src/users/mod.rs
View file

@ -1,3 +1,4 @@
pub mod contact;
pub mod user_community_util;
pub mod user_manager;
pub mod user_profile;

58
iota-storage/src/users/user_community_util.rs Normal file
View file

@ -0,0 +1,58 @@
use iota_util::file_util::{load_file, save_file};
use json::{self, Array, JsonValue};
pub struct UserCommunityUtil;
impl UserCommunityUtil {
pub fn add_community(storage_owner: i64, address: String, title: String, position: String) {
let file_path = format!("users/{}/", storage_owner);
let mut communities = Self::load_array(&file_path, "communities.json");
let mut community = JsonValue::new_object();
community["title"] = JsonValue::String(title);
community["address"] = JsonValue::String(address);
community["position"] = JsonValue::String(position);
communities.push(community);
save_file(
&file_path,
"communities.json",
&JsonValue::Array(communities).to_string(),
);
}
pub fn remove_community(storage_owner: i64, community_address: String) {
let file_path = format!("users/{}/", storage_owner);
let communities = Self::load_array(&file_path, "communities.json");
let filtered: Array = communities
.iter()
.filter(|entry| entry["address"].as_str() != Some(&community_address))
.cloned()
.collect();
save_file(
&file_path,
"communities.json",
&JsonValue::Array(filtered).to_string(),
);
}
pub fn get_communities(storage_owner: i64) -> Array {
let file_path = format!("users/{}/", storage_owner);
Self::load_array(&file_path, "communities.json")
}
fn load_array(dir: &str, name: &str) -> Array {
let content = load_file(dir, name);
if content.is_empty() {
return Array::new();
}
let parsed = json::parse(&content);
match parsed {
Ok(JsonValue::Array(arr)) => arr,
_ => Array::new(),
}
}
}

298
iota-storage/src/users/user_manager.rs
View file

@ -1,214 +1,15 @@
use crate::users::user_profile::UserProfile;
use crate::util::db;
use base64::{Engine as _, engine::general_purpose::STANDARD};
use iota_util::crypto_helper::{self, hex_hash, keyring_from_base64, public_key_bundle_to_base64};
use iota_util::file_util::{load_file, save_file};
use json::JsonValue;
use once_cell::sync::Lazy;
use rand_core::{OsRng, RngCore};
use rusqlite::params;
use std::io::{self};
use std::sync::Mutex;
pub fn add_user(user: UserProfile) {
if let Err(e) = db::with_db(|conn| {
conn.execute(
r#"
INSERT INTO users (user_id, username, public_key, private_key_hash, reset_token, created_at, display_name)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
ON CONFLICT(user_id) DO UPDATE SET
username = excluded.username,
public_key = excluded.public_key,
private_key_hash = excluded.private_key_hash,
reset_token = excluded.reset_token,
display_name = excluded.display_name
"#,
params![
user.user_id,
user.username,
user.public_key,
user.private_key_hash,
user.reset_token,
user.created_at,
user.display_name,
],
)?;
for (app_id, app_secret) in &user.trusted_apps {
conn.execute(
r#"
INSERT OR REPLACE INTO trusted_apps (user_id, app_id, app_secret)
VALUES (?1, ?2, ?3)
"#,
params![user.user_id, app_id, app_secret],
)?;
}
Ok(())
}) {
eprintln!("Failed to add_user: {}", e);
}
}
pub fn update_user(user: UserProfile) {
add_user(user);
}
pub fn get_user_by_username(username: &str) -> Option<UserProfile> {
match db::with_db(|conn| {
match conn.query_row(
"SELECT user_id, username, public_key, private_key_hash, reset_token, created_at, display_name FROM users WHERE username = ?1 LIMIT 1",
params![username],
|r| {
let user_id: i64 = r.get(0)?;
Ok(UserProfile {
user_id,
username: r.get(1)?,
display_name: r.get(6)?,
public_key: r.get(2)?,
private_key_hash: r.get(3)?,
created_at: r.get(5)?,
reset_token: r.get(4)?,
trusted_apps: load_trusted_apps(user_id),
})
},
) {
Ok(user) => Ok(Some(user)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}) {
Ok(opt) => opt,
Err(e) => {
eprintln!("Error querying user by username: {}", e);
None
}
}
}
pub fn get_user(user_id: i64) -> Option<UserProfile> {
match db::with_db(|conn| {
match conn.query_row(
"SELECT user_id, username, public_key, private_key_hash, reset_token, created_at, display_name FROM users WHERE user_id = ?1 LIMIT 1",
params![user_id],
|r| {
let user_id: i64 = r.get(0)?;
Ok(UserProfile {
user_id,
username: r.get(1)?,
display_name: r.get(6)?,
public_key: r.get(2)?,
private_key_hash: r.get(3)?,
created_at: r.get(5)?,
reset_token: r.get(4)?,
trusted_apps: load_trusted_apps(user_id),
})
},
) {
Ok(user) => Ok(Some(user)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}) {
Ok(opt) => opt,
Err(e) => {
eprintln!("Error querying user: {}", e);
None
}
}
}
pub fn get_users() -> Vec<UserProfile> {
match db::with_db(|conn| {
let mut stmt = conn.prepare(
r#"
SELECT user_id, username, public_key, private_key_hash, reset_token, created_at, display_name
FROM users
ORDER BY username
"#,
)?;
let rows = stmt.query_map([], |r| {
let user_id: i64 = r.get(0)?;
let username: String = r.get(1)?;
let public_key: String = r.get(2)?;
let private_key_hash: String = r.get(3)?;
let reset_token: String = r.get(4)?;
let created_at: i64 = r.get(5)?;
let display_name: Option<String> = r.get(6)?;
Ok(UserProfile {
user_id,
username,
display_name,
public_key,
private_key_hash,
created_at,
reset_token,
trusted_apps: std::collections::HashMap::new(),
})
})?;
let mut out = Vec::new();
for row in rows {
match row {
Ok(mut user) => {
user.trusted_apps = load_trusted_apps(user.user_id);
out.push(user);
}
Err(e) => eprintln!("Failed to read user row: {}", e),
}
}
Ok(out)
}) {
Ok(v) => v,
Err(e) => {
eprintln!("Failed to query users: {}", e);
Vec::new()
}
}
}
fn load_trusted_apps(user_id: i64) -> std::collections::HashMap<String, String> {
match db::with_db(|conn| {
let mut stmt =
conn.prepare("SELECT app_id, app_secret FROM trusted_apps WHERE user_id = ?1")?;
let rows = stmt.query_map(params![user_id], |r| {
Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
})?;
let mut map = std::collections::HashMap::new();
for row in rows {
if let Ok((k, v)) = row {
map.insert(k, v);
}
}
Ok(map)
}) {
Ok(m) => m,
Err(e) => {
eprintln!("Failed to load trusted apps: {}", e);
std::collections::HashMap::new()
}
}
}
pub fn remove_user(user_id: i64) {
if let Err(e) = db::with_db(|conn| {
conn.execute(
"DELETE FROM trusted_apps WHERE user_id = ?1",
params![user_id],
)?;
conn.execute("DELETE FROM users WHERE user_id = ?1", params![user_id])?;
Ok(())
}) {
eprintln!("Failed to remove_user: {}", e);
}
}
pub fn clear() {
if let Err(e) = db::with_db(|conn| {
conn.execute_batch("DELETE FROM trusted_apps; DELETE FROM users;")?;
Ok(())
}) {
eprintln!("Failed to clear users: {}", e);
}
}
static USERS: Lazy<Mutex<Vec<UserProfile>>> = Lazy::new(|| Mutex::new(Vec::new()));
static UNIQUE: Lazy<Mutex<bool>> = Lazy::new(|| Mutex::new(false));
#[allow(dead_code)]
pub async fn load_from_tu(username: &str) -> Result<(), ()> {
@ -234,38 +35,91 @@ pub async fn load_from_tu(username: &str) -> Result<(), ()> {
hex_hash(&keyring_b64),
reset_token,
);
add_user(user_profile);
USERS.lock().unwrap().push(user_profile);
Ok(())
}
pub fn save_users() {
// No-op: users are auto-saved via SQLite.
pub fn add_user(user: UserProfile) {
USERS.lock().unwrap().push(user);
}
pub async fn load_users() -> std::io::Result<()> {
// Users are loaded from SQLite on demand. This function is kept for API compat.
// If we need to migrate from a legacy users.json file, we can do so here.
pub fn update_user(user: UserProfile) {
let mut users = USERS.lock().unwrap();
if let Some(pos) = users.iter().position(|u| u.user_id == user.user_id) {
users[pos] = user;
}
*UNIQUE.lock().unwrap() = true;
}
pub fn get_user_by_username(username: &str) -> Option<UserProfile> {
USERS
.lock()
.unwrap()
.iter()
.cloned()
.find(|u| u.username == username)
}
pub fn get_user(user_id: i64) -> Option<UserProfile> {
USERS
.lock()
.unwrap()
.iter()
.cloned()
.find(|u| u.user_id == user_id)
}
pub fn get_users() -> Vec<UserProfile> {
USERS.lock().unwrap().clone()
}
pub fn remove_user(user_id: i64) {
let mut users = USERS.lock().unwrap();
users.retain(|u| u.user_id != user_id);
*UNIQUE.lock().unwrap() = true;
}
pub fn save_users() {
*UNIQUE.lock().unwrap() = false;
let users = USERS.lock().unwrap();
let arr: Vec<JsonValue> = users.iter().map(|u| u.to_json()).collect();
let json_str = JsonValue::Array(arr).dump();
save_file("", "users.json", &json_str);
}
pub fn clear() {
let mut users = USERS.lock().unwrap();
users.clear();
*UNIQUE.lock().unwrap() = true;
}
pub async fn load_users() -> io::Result<()> {
let content = load_file("", "users.json");
if content.trim().is_empty() {
return Ok(());
}
if let Ok(parsed) = json::parse(&content) {
if let json::JsonValue::Array(arr) = parsed {
for j in arr.iter() {
if let Some(up) = UserProfile::from_json(j) {
add_user(up);
}
let parsed =
json::parse(&content).map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?;
if let JsonValue::Array(arr) = parsed {
let mut users = USERS.lock().unwrap();
for j in arr.iter() {
if let Some(up) = UserProfile::from_json(j).await {
users.push(up);
}
}
}
// Rename the old file so we don't re-import
let _ = std::fs::rename(
std::path::PathBuf::from(iota_util::file_util::get_directory()).join("users.json"),
std::path::PathBuf::from(iota_util::file_util::get_directory()).join("users.json.imported"),
);
if *UNIQUE.lock().unwrap() {
save_users();
}
Ok(())
}
#[allow(dead_code)]
pub fn set_unique(val: bool) {
*UNIQUE.lock().unwrap() = val;
}
pub fn save_app_data(user_id: i64, app_identifier: &str, data: &str) {
let path = format!("users/{}/apps", user_id);
let name = format!("{}.json", app_identifier);

52
iota-storage/src/users/user_profile.rs
View file

@ -5,9 +5,9 @@ use iota_util::file_util::{has_file, load_file, used_dir_space};
use json::{JsonValue, object};
use rand::Rng;
use rand::rngs::OsRng;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, Serialize, Deserialize)]
// --- UserProfile ---
#[derive(Clone, Debug)]
pub struct UserProfile {
pub user_id: i64,
pub username: String,
@ -43,6 +43,26 @@ impl UserProfile {
}
}
pub fn to_json(&self) -> JsonValue {
let mut trusted_apps_obj = json::JsonValue::new_object();
for (k, v) in &self.trusted_apps {
trusted_apps_obj[k] = v.clone().into();
}
let mut obj = object! {
"uuid" => self.user_id,
"username" => self.username.clone(),
"public_key" => self.public_key.clone(),
"private_key_hash" => self.private_key_hash.clone(),
"created_at" => self.created_at,
"reset_token" => self.reset_token.clone(),
"trusted_apps" => trusted_apps_obj,
};
if let Some(d) = &self.display_name {
obj["display_name"] = d.clone().into();
}
obj
}
pub fn frontend(&self) -> JsonValue {
let mut obj = object! {
"uuid" => self.user_id,
@ -58,11 +78,10 @@ impl UserProfile {
if has_file("", &format!("{}.tu", self.username.clone())) {
obj["tu"] = load_file("", &format!("{}.tu", self.username.clone())).into();
}
obj
}
/// Legacy JSON import - used when migrating from users.json to SQLite.
pub fn from_json(j: &JsonValue) -> Option<Self> {
pub async fn from_json(j: &JsonValue) -> Option<Self> {
let user_id = j["uuid"].as_i64()?;
let username = j["username"].as_str()?.to_string();
let public_key = j["public_key"].as_str()?.to_string();
@ -80,7 +99,7 @@ impl UserProfile {
}
}
Some(UserProfile {
let up = UserProfile {
user_id,
username,
display_name,
@ -89,15 +108,22 @@ impl UserProfile {
created_at,
reset_token,
trusted_apps,
})
}
};
pub fn from_yaml(s: &str) -> Result<Self, serde_yaml::Error> {
serde_yaml::from_str(s)
}
// TODO: Migrate to Omikron / Wss
/* if j.has_key("migrate")
|| j.has_key("migrating")
|| j.has_key("changing")
|| j.has_key("move")
|| j.has_key("moving")
{
if auth_connector::migrate_user(&mut up).await {
log_message(format!("[INFO] Migration triggered for {}", up.username));
user_manager::set_unique(true);
}
} */
pub fn to_yaml(&self) -> Result<String, serde_yaml::Error> {
serde_yaml::to_string(self)
Some(up)
}
#[allow(dead_code)]

494
iota-storage/src/util/chat_files.rs
View file

@ -1,7 +1,9 @@
use crate::storage_error::StorageError;
use crate::util::db;
use iota_logger::log;
use json::{JsonValue, array, object};
use rusqlite::params;
use std::io;
use std::sync::{Arc, LazyLock, Mutex};
#[derive(PartialEq, Debug, Clone)]
pub enum MessageState {
@ -43,327 +45,11 @@ impl MessageState {
}
}
#[derive(Debug, Clone)]
pub struct StoredMessage {
pub id: i64,
pub message_time: i64,
pub content: String,
pub edited: bool,
pub sent_by_self: bool,
pub message_state: String,
pub height: i64,
pub reply_to: Option<i64>,
pub reactions: Vec<StoredReaction>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredReaction {
pub reaction: String,
pub user_id: i64,
}
/*
* Each edit is recorded in message_edits with the before/after content and a
* timestamp. Only the original sender (sent_by_self = 1) may edit.
*/
pub fn edit_message(
storage_owner: i64,
external_user: i64,
message_time: i64,
editor_id: i64,
new_content: &str,
) -> Result<(), StorageError> {
update_message_content(
storage_owner,
external_user,
message_time,
editor_id,
new_content,
true,
)
}
/* Applies an edit received from the message sender to the recipient's copy. */
pub fn apply_remote_edit(
storage_owner: i64,
external_user: i64,
message_time: i64,
editor_id: i64,
new_content: &str,
) -> Result<(), StorageError> {
if editor_id != external_user {
return Err(StorageError::Other(
"Remote editor does not match chat partner".into(),
));
}
update_message_content(
storage_owner,
external_user,
message_time,
editor_id,
new_content,
false,
)
}
fn update_message_content(
storage_owner: i64,
external_user: i64,
message_time: i64,
editor_id: i64,
new_content: &str,
require_sent_by_self: bool,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let msg = conn.query_row(
r#"
SELECT id, content, sent_by_self
FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
"#,
params![storage_owner, external_user, message_time],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, i64>(2)?,
))
},
)?;
let (msg_id, old_content, sent_by_self) = msg;
if require_sent_by_self && sent_by_self != 1 {
return Err(StorageError::Other(
"Only the original sender can edit this message".into(),
));
}
if !require_sent_by_self && sent_by_self != 0 {
return Err(StorageError::Other(
"Remote edits may only update received messages".into(),
));
}
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
conn.execute(
r#"
INSERT INTO message_edits (message_id, content_before, content_after, edited_at, edited_by)
VALUES (?1, ?2, ?3, ?4, ?5)
"#,
params![msg_id, old_content, new_content, now, editor_id],
)?;
conn.execute(
r#"
UPDATE messages
SET content = ?1, edited_count = edited_count + 1
WHERE id = ?2
"#,
params![new_content, msg_id],
)?;
Ok(())
})
}
pub fn hard_delete_message(
storage_owner: i64,
external_user: i64,
message_time: i64,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let msg_id: i64 = conn.query_row(
r#"
SELECT id FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
"#,
params![storage_owner, external_user, message_time],
|row| row.get(0),
)?;
conn.execute(
"DELETE FROM message_edits WHERE message_id = ?1",
params![msg_id],
)?;
conn.execute(
"DELETE FROM reactions WHERE message_id = ?1",
params![msg_id],
)?;
conn.execute("DELETE FROM messages WHERE id = ?1", params![msg_id])?;
Ok(())
})
}
/* Deletes a message from the sender's local copy after checking ownership. */
pub fn delete_message(
storage_owner: i64,
external_user: i64,
message_time: i64,
) -> Result<(), StorageError> {
ensure_message_direction(storage_owner, external_user, message_time, true)?;
hard_delete_message(storage_owner, external_user, message_time)
}
/* Flags the recipient's local copy after validating its sender, preserving its history. */
pub fn apply_remote_delete(
storage_owner: i64,
external_user: i64,
message_time: i64,
sender_id: i64,
) -> Result<(), StorageError> {
if sender_id != external_user {
return Err(StorageError::Other(
"Remote sender does not match chat partner".into(),
));
}
ensure_message_direction(storage_owner, external_user, message_time, false)?;
flag_deleted_by_external(storage_owner, external_user, message_time)
}
fn ensure_message_direction(
storage_owner: i64,
external_user: i64,
message_time: i64,
expected_sent_by_self: bool,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let sent_by_self: i64 = conn.query_row(
r#"
SELECT sent_by_self FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
"#,
params![storage_owner, external_user, message_time],
|row| row.get(0),
)?;
if (sent_by_self != 0) != expected_sent_by_self {
return Err(StorageError::Other(
"Message sender is not authorized".into(),
));
}
Ok(())
})
}
/*
* Marks a message as deleted by the external user rather than removing the row,
* so the storage owner still sees a tombstone in the UI.
*/
pub fn flag_deleted_by_external(
storage_owner: i64,
external_user: i64,
message_time: i64,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let affected = conn.execute(
r#"
UPDATE messages
SET deleted_by_external = 1
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
"#,
params![storage_owner, external_user, message_time],
)?;
if affected == 0 {
return Err(StorageError::Other("Message not found".into()));
}
Ok(())
})
}
/*
* Removes the edit trail but keeps the message with edited_count > 0 so
* the UI still shows the "edited" indicator. Only the own user should
* call this.
*/
pub fn delete_edit_history(
storage_owner: i64,
external_user: i64,
message_time: i64,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let msg_id: i64 = conn.query_row(
r#"
SELECT id FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
"#,
params![storage_owner, external_user, message_time],
|row| row.get(0),
)?;
conn.execute(
"DELETE FROM message_edits WHERE message_id = ?1",
params![msg_id],
)?;
Ok(())
})
}
pub fn add_reaction(
storage_owner: i64,
external_user: i64,
message_time: i64,
user_id: i64,
reaction: &str,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let msg_id: i64 = conn.query_row(
r#"
SELECT id FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
"#,
params![storage_owner, external_user, message_time],
|row| row.get(0),
)?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
conn.execute(
r#"
INSERT OR IGNORE INTO reactions (message_id, user_id, reaction, created_at)
VALUES (?1, ?2, ?3, ?4)
"#,
params![msg_id, user_id, reaction, now],
)?;
Ok(())
})
}
pub fn remove_reaction(
storage_owner: i64,
external_user: i64,
message_time: i64,
user_id: i64,
reaction: &str,
) -> Result<(), StorageError> {
db::with_db(|conn| {
let msg_id: i64 = conn.query_row(
r#"
SELECT id FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
"#,
params![storage_owner, external_user, message_time],
|row| row.get(0),
)?;
conn.execute(
"DELETE FROM reactions WHERE message_id = ?1 AND user_id = ?2 AND reaction = ?3",
params![msg_id, user_id, reaction],
)?;
Ok(())
})
}
// Shared DB created via helper.
// The db helper constructs the messages sqlite file and ensures PRAGMAs and schema exist.
static MESSAGES_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
db::create_general_messages_db().expect("Failed to create or initialize general messages DB")
});
pub fn add_message(
send_time: u128,
@ -382,12 +68,19 @@ pub fn add_message(
}
};
if let Err(e) = db::with_db(|conn| {
// Insert the message into the DB
let insert_result = db::with_conn(&MESSAGES_DB, |conn| {
conn.execute(
r#"
INSERT INTO messages (
storage_owner, external_user, message_time, content,
sent_by_self, message_state, height, reply_to
storage_owner,
external_user,
message_time,
content,
sent_by_self,
message_state,
height,
reply_to
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
"#,
params![
@ -406,13 +99,18 @@ pub fn add_message(
],
)?;
Ok(())
}) {
});
if let Err(e) = insert_result {
log!("Failed to insert message into sqlite: {}", e);
return;
}
// Update contacts table to reflect that this conversation exists and has a recent message.
// Use the Contact helper to set last_message_at to the message timestamp.
let mut contact = crate::users::contact::Contact::new(external_user);
contact.set_last_message_at(message_time);
// This will insert or update the contact for the storage owner.
crate::util::chats_util::mod_user(storage_owner, &contact);
}
@ -421,21 +119,25 @@ pub fn change_message_state(
storage_owner: i64,
external_user: i64,
new_state: MessageState,
) -> std::io::Result<()> {
db::with_db(|conn| {
) -> io::Result<()> {
// Run the SELECT and UPDATE inside with_conn to centralize connection access.
let res: Result<(), String> = db::with_conn(&MESSAGES_DB, |conn| {
let current: Option<String> = match conn.query_row(
r#"
SELECT message_state
FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
ORDER BY id DESC LIMIT 1
WHERE storage_owner = ?1
AND external_user = ?2
AND message_time = ?3
ORDER BY id DESC
LIMIT 1
"#,
params![storage_owner, external_user, timestamp],
|row| row.get(0),
) {
Ok(state) => Some(state),
Err(rusqlite::Error::QueryReturnedNoRows) => None,
Err(e) => return Err(e.into()),
Err(e) => return Err(e),
};
let Some(current_state_raw) = current else {
@ -452,58 +154,24 @@ pub fn change_message_state(
UPDATE messages
SET message_state = ?1
WHERE id = (
SELECT id FROM messages
WHERE storage_owner = ?2 AND external_user = ?3 AND message_time = ?4
ORDER BY id DESC LIMIT 1
SELECT id
FROM messages
WHERE storage_owner = ?2
AND external_user = ?3
AND message_time = ?4
ORDER BY id DESC
LIMIT 1
)
"#,
params![upgraded, storage_owner, external_user, timestamp],
)?;
Ok(())
})
.map_err(|e: StorageError| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))
}
});
fn load_reactions(
conn: &rusqlite::Connection,
msg_ids: &[i64],
) -> std::collections::HashMap<i64, Vec<StoredReaction>> {
if msg_ids.is_empty() {
return std::collections::HashMap::new();
match res {
Ok(_) => Ok(()),
Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)),
}
let placeholders: Vec<String> = msg_ids
.iter()
.enumerate()
.map(|(i, _)| format!("?{}", i + 1))
.collect();
let query = format!(
"SELECT message_id, reaction, user_id FROM reactions WHERE message_id IN ({}) ORDER BY created_at ASC, id ASC",
placeholders.join(", ")
);
let mut map: std::collections::HashMap<i64, Vec<StoredReaction>> =
std::collections::HashMap::new();
if let Ok(mut stmt) = conn.prepare(&query) {
let params: Vec<&dyn rusqlite::types::ToSql> = msg_ids
.iter()
.map(|id| id as &dyn rusqlite::types::ToSql)
.collect();
if let Ok(rows) = stmt.query_map(params.as_slice(), |row| {
Ok((
row.get::<_, i64>(0)?,
StoredReaction {
reaction: row.get(1)?,
user_id: row.get(2)?,
},
))
}) {
for row in rows.flatten() {
map.entry(row.0).or_default().push(row.1);
}
}
}
map
}
pub fn get_messages(
@ -511,17 +179,26 @@ pub fn get_messages(
external_user: i64,
loaded_messages: i64,
amount: i64,
) -> Vec<StoredMessage> {
) -> JsonValue {
let messages = array![];
if amount <= 0 || loaded_messages < 0 {
return Vec::new();
return messages;
}
match db::with_db(|conn| {
let res: Result<JsonValue, String> = db::with_conn(&MESSAGES_DB, |conn| {
let mut stmt = conn.prepare(
r#"
SELECT id, message_time, content, sent_by_self, message_state, height, reply_to, edited_count
SELECT
message_time,
content,
sent_by_self,
message_state,
height,
reply_to
FROM messages
WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0
WHERE storage_owner = ?1
AND external_user = ?2
ORDER BY message_time DESC, id DESC
LIMIT ?3 OFFSET ?4
"#,
@ -530,40 +207,49 @@ pub fn get_messages(
let rows = stmt.query_map(
params![storage_owner, external_user, amount, loaded_messages],
|row| {
Ok(StoredMessage {
id: row.get(0)?,
message_time: row.get(1)?,
content: row.get(2)?,
sent_by_self: row.get::<_, i64>(3)? != 0,
message_state: row.get(4)?,
height: row.get(5).unwrap_or(0),
reply_to: row.get(6).ok().flatten(),
edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
reactions: Vec::new(),
})
let message_time: i64 = row.get(0)?;
let content: String = row.get(1)?;
let sent_by_self: i64 = row.get(2)?;
let message_state: String = row.get(3)?;
let height: i64 = row.get(4).unwrap_or(0);
let reply_to: Option<i64> = row.get(5).ok().flatten();
Ok((message_time, content, sent_by_self, message_state, height, reply_to))
},
)?;
let mut out = Vec::new();
let mut out = array![];
for row in rows {
match row {
Ok(msg) => out.push(msg),
Err(e) => log!("Failed to read row from sqlite: {}", e),
Ok((message_time, content, sent_by_self, message_state, height, reply_to)) => {
let mut msg = object! {
"message_time" => message_time,
"content" => content,
"sent_by_self" => (sent_by_self != 0),
"message_state" => message_state,
"height" => height
};
if let Some(rt) = reply_to {
let _ = msg.insert("reply_to", rt);
}
if let Err(e) = out.push(msg) {
// out.push returns a JsonError; log it instead of using `?` to avoid
// incompatible error conversions inside the DB closure.
log!("Failed to append message to output array: {:?}", e);
}
}
Err(e) => {
log!("Failed to read row from sqlite: {}", e);
}
}
}
let msg_ids: Vec<i64> = out.iter().map(|m| m.id).collect();
let reaction_map = load_reactions(conn, &msg_ids);
for msg in &mut out {
msg.reactions = reaction_map.get(&msg.id).cloned().unwrap_or_default();
}
Ok(out)
}) {
});
match res {
Ok(v) => v,
Err(e) => {
log!("Failed to query messages: {}", e);
Vec::new()
messages
}
}
}

57
iota-storage/src/util/chats_util.rs
View file

@ -1,13 +1,24 @@
use crate::users::contact::Contact;
use crate::util::db;
use rusqlite::params;
use std::sync::{Arc, LazyLock, Mutex};
/// Shared DB connection for contacts/messages (created by db helper).
static MESSAGES_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
db::create_general_messages_db().expect("Failed to create or initialize general messages DB")
});
/// Insert or update a contact for the given storage owner.
pub fn mod_user(storage_owner: i64, contact: &Contact) {
if let Err(e) = db::with_db(|conn| {
if let Err(e) = db::with_conn(&MESSAGES_DB, |conn| {
conn.execute(
r#"
INSERT INTO contacts (storage_owner, user_id, user_name, last_message_at)
VALUES (?1, ?2, ?3, ?4)
INSERT INTO contacts (
storage_owner,
user_id,
user_name,
last_message_at
) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(storage_owner, user_id) DO UPDATE SET
user_name = excluded.user_name,
last_message_at = excluded.last_message_at
@ -16,7 +27,7 @@ pub fn mod_user(storage_owner: i64, contact: &Contact) {
storage_owner,
contact.user_id,
contact.user_name.clone(),
contact.last_message_at,
contact.last_message_at
],
)?;
Ok(())
@ -25,8 +36,9 @@ pub fn mod_user(storage_owner: i64, contact: &Contact) {
}
}
/// Retrieve a single contact for storage_owner/user_id.
pub fn get_user(storage_owner: i64, user_id: i64) -> Option<Contact> {
match db::with_db(|conn| {
let res: Result<Option<Contact>, String> = db::with_conn(&MESSAGES_DB, |conn| {
match conn.query_row(
r#"
SELECT user_id, user_name, last_message_at
@ -36,18 +48,23 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Option<Contact> {
"#,
params![storage_owner, user_id],
|r| {
let user_id: i64 = r.get(0)?;
let user_name: Option<String> = r.get(1)?;
let last_message_at: Option<i64> = r.get(2)?;
Ok(Contact {
user_id: r.get(0)?,
user_name: r.get(1)?,
last_message_at: r.get(2)?,
user_id,
user_name,
last_message_at,
})
},
) {
Ok(c) => Ok(Some(c)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
Err(e) => Err(e),
}
}) {
});
match res {
Ok(opt) => opt,
Err(e) => {
eprintln!("Error querying user in get_user: {}", e);
@ -56,8 +73,11 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Option<Contact> {
}
}
/// Retrieve all contacts for a storage owner, ordered by last_message_at desc / user_id asc.
pub fn get_users(storage_owner: i64) -> Vec<Contact> {
match db::with_db(|conn| {
let contacts_out = Vec::new();
let res: Result<Vec<Contact>, String> = db::with_conn(&MESSAGES_DB, |conn| {
let mut stmt = conn.prepare(
r#"
SELECT user_id, user_name, last_message_at
@ -71,10 +91,13 @@ pub fn get_users(storage_owner: i64) -> Vec<Contact> {
)?;
let rows = stmt.query_map(params![storage_owner], |r| {
let user_id: i64 = r.get(0)?;
let user_name: Option<String> = r.get(1)?;
let last_message_at: Option<i64> = r.get(2)?;
Ok(Contact {
user_id: r.get(0)?,
user_name: r.get(1)?,
last_message_at: r.get(2)?,
user_id,
user_name,
last_message_at,
})
})?;
@ -86,11 +109,13 @@ pub fn get_users(storage_owner: i64) -> Vec<Contact> {
}
}
Ok(out)
}) {
});
match res {
Ok(v) => v,
Err(e) => {
eprintln!("Failed to query contacts in get_users: {}", e);
Vec::new()
contacts_out
}
}
}

56
iota-storage/src/util/communities_util.rs
View file

@ -1,22 +1,25 @@
use crate::util::db;
use json::Array;
use rusqlite::params;
use std::sync::{Arc, LazyLock, Mutex};
#[derive(Debug, Clone)]
pub struct StoredCommunity {
pub address: String,
pub title: String,
pub position: String,
}
static MESSAGES_DB: LazyLock<Arc<Mutex<rusqlite::Connection>>> = LazyLock::new(|| {
db::create_general_messages_db().expect("Failed to create or initialize general messages DB")
});
pub struct CommunitiesUtil;
impl CommunitiesUtil {
pub fn add_community(storage_owner: i64, address: String, title: String, position: String) {
if let Err(e) = db::with_db(|conn| {
if let Err(e) = db::with_conn(&MESSAGES_DB, |conn| {
conn.execute(
r#"
INSERT INTO communities (storage_owner, address, title, position)
VALUES (?1, ?2, ?3, ?4)
INSERT INTO communities (
storage_owner,
address,
title,
position
) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(storage_owner, address) DO UPDATE SET
title = excluded.title,
position = excluded.position
@ -30,7 +33,7 @@ impl CommunitiesUtil {
}
pub fn remove_community(storage_owner: i64, community_address: String) {
if let Err(e) = db::with_db(|conn| {
if let Err(e) = db::with_conn(&MESSAGES_DB, |conn| {
conn.execute(
"DELETE FROM communities WHERE storage_owner = ?1 AND address = ?2",
params![storage_owner, community_address],
@ -41,8 +44,10 @@ impl CommunitiesUtil {
}
}
pub fn get_communities(storage_owner: i64) -> Vec<StoredCommunity> {
match db::with_db(|conn| {
pub fn get_communities(storage_owner: i64) -> Array {
let communities_out = Array::new();
let res: Result<Array, String> = db::with_conn(&MESSAGES_DB, |conn| {
let mut stmt = conn.prepare(
r#"
SELECT address, title, position
@ -52,26 +57,33 @@ impl CommunitiesUtil {
)?;
let rows = stmt.query_map(params![storage_owner], |r| {
Ok(StoredCommunity {
address: r.get(0)?,
title: r.get(1)?,
position: r.get(2)?,
})
let address: String = r.get(0)?;
let title: String = r.get(1)?;
let position: String = r.get(2)?;
Ok((address, title, position))
})?;
let mut out = Vec::new();
let mut out = Array::new();
for row in rows {
match row {
Ok(community) => out.push(community),
Ok((address, title, position)) => {
let mut community = json::JsonValue::new_object();
community["title"] = json::JsonValue::String(title);
community["address"] = json::JsonValue::String(address);
community["position"] = json::JsonValue::String(position);
out.push(community);
}
Err(e) => eprintln!("Failed to read community row: {}", e),
}
}
Ok(out)
}) {
Ok(v) => v,
});
match res {
Ok(arr) => arr,
Err(e) => {
eprintln!("Failed to query communities in get_communities: {}", e);
Vec::new()
communities_out
}
}
}

10
iota-storage/src/util/config_util.rs
View file

@ -51,17 +51,17 @@ impl Default for IotaConfig {
}
pub fn load_config() {
let s = load_file("", "config.yaml");
let s = load_file("", "config.json");
if s.is_empty() {
return;
}
match serde_yaml::from_str::<IotaConfig>(&s) {
match serde_json::from_str::<IotaConfig>(&s) {
Ok(parsed) => {
CONFIG.store(Arc::new(parsed));
}
Err(e) => {
eprintln!("Failed to parse config.yaml: {}. Content: '{}'", e, s);
eprintln!("Failed to parse config.json: {}. Content: '{}'", e, s);
}
}
}
@ -72,8 +72,8 @@ pub fn clear_config() {
}
pub fn save_config() {
if let Ok(yaml) = serde_yaml::to_string(&**CONFIG.load()) {
save_file("", "config.yaml", &yaml);
if let Ok(json) = serde_json::to_string(&**CONFIG.load()) {
save_file("", "config.json", &json);
}
}

410
iota-storage/src/util/db.rs
View file

@ -1,247 +1,88 @@
//! Database helper utilities.
//!
//! This module provides small helpers to open/init sqlite databases and to
//! create a shared (Arc<Mutex<Connection>>) connection wrapper callers can
//! reuse. The goal is to centralize the "open and initialize" logic and
//! provide small convenience helpers used by other util modules.
use iota_util::file_util::get_directory;
use once_cell::sync::Lazy;
use r2d2::ManageConnection;
use rusqlite::Connection;
use rusqlite::{Connection, Error as RusqliteError};
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::storage_error::StorageError;
const DB_NAME: &str = "messages";
/// A simple r2d2 manager for rusqlite connections.
pub struct SqliteManager;
impl ManageConnection for SqliteManager {
type Connection = Connection;
type Error = rusqlite::Error;
fn connect(&self) -> Result<Connection, rusqlite::Error> {
let path = db_file_path(DB_NAME);
let conn = Connection::open(path)?;
conn.execute_batch("PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL;")?;
conn.busy_timeout(Duration::from_millis(250))?;
Ok(conn)
}
fn is_valid(&self, conn: &mut Connection) -> Result<(), rusqlite::Error> {
conn.execute_batch("SELECT 1")
}
fn has_broken(&self, _conn: &mut Connection) -> bool {
false
}
}
static POOL: Lazy<Arc<r2d2::Pool<SqliteManager>>> = Lazy::new(|| {
let manager = SqliteManager;
let pool = r2d2::Pool::builder()
.max_size(8)
.build(manager)
.expect("Failed to create database connection pool");
run_migrations(&pool).expect("Failed to run database migrations");
Arc::new(pool)
});
pub fn pool() -> Arc<r2d2::Pool<SqliteManager>> {
POOL.clone()
}
pub fn with_db<T, F>(f: F) -> Result<T, StorageError>
where
F: FnOnce(&Connection) -> Result<T, StorageError>,
{
let conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?;
f(&conn)
}
fn db_file_path(db_name: &str) -> String {
/// Returns the file path for a named DB inside the application's data directory.
///
/// Arguments:
/// - `db_name` : name of the DB (without extension). Example: `"messages"`.
pub fn db_file_path(db_name: &str) -> String {
let mut p = PathBuf::from(get_directory());
p.push(format!("{db_name}.sqlite3"));
p.to_string_lossy().to_string()
}
fn run_migrations(pool: &r2d2::Pool<SqliteManager>) -> Result<(), StorageError> {
let conn = pool.get().map_err(|e| StorageError::Pool(e.to_string()))?;
run_migrations_on_connection(&conn)
}
/*
* Older builds could apply a schema change without advancing user_version.
* Check each added column so those databases can resume upgrading.
*/
fn add_column_if_missing(
conn: &Connection,
column: &str,
definition: &str,
) -> Result<(), StorageError> {
let mut statement =
conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
let exists = statement.exists([column])?;
if !exists {
conn.execute_batch(&format!("ALTER TABLE messages ADD COLUMN {definition};"))?;
}
Ok(())
}
fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
let current_version: i64 = conn
.pragma_query_value(None, "user_version", |r| r.get(0))
.unwrap_or(0);
if current_version < 1 {
conn.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
external_user INTEGER NOT NULL,
message_time INTEGER NOT NULL,
content TEXT NOT NULL,
sent_by_self INTEGER NOT NULL,
message_state TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_lookup
ON messages (storage_owner, external_user, message_time DESC);
CREATE TABLE IF NOT EXISTS contacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
user_id INTEGER NOT NULL,
user_name TEXT,
last_message_at INTEGER,
UNIQUE(storage_owner, user_id)
);
CREATE INDEX IF NOT EXISTS idx_contacts_owner
ON contacts (storage_owner, last_message_at DESC, user_id ASC);
CREATE TABLE IF NOT EXISTS communities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
address TEXT NOT NULL,
title TEXT NOT NULL,
position TEXT NOT NULL,
UNIQUE(storage_owner, address)
);
CREATE INDEX IF NOT EXISTS idx_communities_owner
ON communities (storage_owner);
CREATE TABLE IF NOT EXISTS users (
user_id INTEGER PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
public_key TEXT NOT NULL,
private_key_hash TEXT NOT NULL,
reset_token TEXT NOT NULL,
created_at INTEGER NOT NULL,
display_name TEXT
);
CREATE TABLE IF NOT EXISTS trusted_apps (
user_id INTEGER NOT NULL,
app_id TEXT NOT NULL,
app_secret TEXT NOT NULL,
PRIMARY KEY (user_id, app_id)
);
PRAGMA user_version = 1;
"#,
)?;
}
if current_version < 2 {
add_column_if_missing(conn, "height", "height INTEGER NOT NULL DEFAULT 0")?;
conn.execute_batch("PRAGMA user_version = 2;")?;
}
if current_version < 3 {
add_column_if_missing(conn, "reply_to", "reply_to INTEGER")?;
conn.execute_batch("PRAGMA user_version = 3;")?;
}
if current_version < 4 {
add_column_if_missing(
conn,
"edited_count",
"edited_count INTEGER NOT NULL DEFAULT 0",
)?;
add_column_if_missing(
conn,
"deleted_by_external",
"deleted_by_external INTEGER NOT NULL DEFAULT 0",
)?;
conn.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS message_edits (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message_id INTEGER NOT NULL REFERENCES messages(id),
content_before TEXT NOT NULL,
content_after TEXT NOT NULL,
edited_at INTEGER NOT NULL,
edited_by INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_message_edits_msg
ON message_edits (message_id, edited_at DESC);
CREATE TABLE IF NOT EXISTS reactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message_id INTEGER NOT NULL REFERENCES messages(id),
user_id INTEGER NOT NULL,
reaction TEXT NOT NULL,
created_at INTEGER NOT NULL,
UNIQUE(message_id, user_id, reaction)
);
CREATE INDEX IF NOT EXISTS idx_reactions_msg
ON reactions (message_id, reaction);
PRAGMA user_version = 4;
"#,
)?;
}
if current_version < 5 {
conn.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS settings (
user_id INTEGER NOT NULL,
session_id INTEGER NOT NULL,
name TEXT NOT NULL,
payload TEXT NOT NULL,
PRIMARY KEY (user_id, session_id, name)
);
CREATE INDEX IF NOT EXISTS idx_settings_lookup
ON settings (user_id, session_id, name);
PRAGMA user_version = 5;
"#,
)?;
}
Ok(())
}
pub fn open_connection(db_name: &str) -> Result<Connection, rusqlite::Error> {
/// Open a sqlite connection to the named DB file (no initialization).
///
/// Arguments:
/// - `db_name`: name of the DB (without extension).
pub fn open_connection(db_name: &str) -> Result<Connection, RusqliteError> {
let path = db_file_path(db_name);
Connection::open(path)
}
/// Open a connection and immediately run `init_sql` via `execute_batch`.
///
/// Arguments:
/// - `db_name`: name of the DB (without extension).
/// - `init_sql`: SQL statements to initialize schema & PRAGMAs (can be multiple).
pub fn open_and_init(db_name: &str, init_sql: &str) -> Result<Connection, RusqliteError> {
let conn = open_connection(db_name)?;
conn.execute_batch(init_sql)?;
Ok(conn)
}
/// Create a shared, Arc<Mutex<Connection>> initialized with the given SQL.
///
/// This is a convenience wrapper that returns an owned Arc<Mutex<Connection>>
/// so caller modules can store it in a `static` or pass it around.
///
/// Arguments:
/// - `db_name`: DB name (without extension).
/// - `init_sql`: init SQL (eg PRAGMA + CREATE TABLE statements).
pub fn create_shared_connection(
db_name: &str,
init_sql: &str,
) -> Result<Arc<std::sync::Mutex<Connection>>, String> {
let path = db_file_path(db_name);
let conn = Connection::open(path).map_err(|e| e.to_string())?;
conn.execute_batch(init_sql).map_err(|e| e.to_string())?;
let _ = conn.busy_timeout(Duration::from_millis(250));
Ok(Arc::new(std::sync::Mutex::new(conn)))
) -> Result<Arc<Mutex<Connection>>, String> {
match open_and_init(db_name, init_sql) {
Ok(conn) => {
// Configure some sensible defaults for concurrency
// Attempt to set a busy timeout to reduce SQLITE_BUSY failures.
let _ = conn.busy_timeout(Duration::from_millis(250));
Ok(Arc::new(Mutex::new(conn)))
}
Err(e) => Err(format!("Failed to open/init DB '{}': {}", db_name, e)),
}
}
pub fn with_conn<T, F>(shared: &Arc<std::sync::Mutex<Connection>>, f: F) -> Result<T, String>
/// Acquire the Connection from an Arc<Mutex<Connection>> and run the provided
/// closure. Converts rusqlite::Error into a String on error.
///
/// Arguments:
/// - `shared`: Arc<Mutex<Connection>>
/// - `f`: closure that receives &Connection and returns Result<T, RusqliteError>
///
/// Returns Ok(T) or Err(String).
pub fn with_conn<T, F>(shared: &Arc<Mutex<Connection>>, f: F) -> Result<T, String>
where
F: FnOnce(&Connection) -> Result<T, rusqlite::Error>,
F: FnOnce(&Connection) -> Result<T, RusqliteError>,
{
// When invoked from within an async runtime (such as Tokio), taking a blocking
// std::sync::Mutex lock on the runtime thread can cause deadlocks or permanent
// awaits. Detect whether we're running inside a Tokio runtime and, if so,
// execute the blocking lock + database closure using Tokio's blocking helper.
//
// The blocking section returns Result<T, String> so we can propagate errors
// in the same form as before.
if tokio::runtime::Handle::try_current().is_ok() {
tokio::task::block_in_place(|| {
let guard = shared
@ -257,44 +98,97 @@ where
}
}
/// Legacy - kept for e2ee_storage which uses its own DB.
pub fn create_general_messages_db() -> Result<Arc<std::sync::Mutex<Connection>>, String> {
create_shared_connection(DB_NAME, "")
}
/// Initialize a general-purpose messages+contacts DB and return a shared
/// connection. This helper creates a single DB file that can contain multiple
/// tables (messages, contacts, ...). The SQL here is conservative and intended
/// to be safe if called multiple times.
///
/// Callers may prefer to call `create_shared_connection("messages", INIT_SQL)`
/// directly, but this convenience is useful for code that expects both tables.
pub fn create_general_messages_db() -> Result<Arc<Mutex<Connection>>, String> {
// Keep PRAGMA and schema in one multi-statement string so callers only
// need to call a single execute_batch.
const INIT_SQL: &str = r#"
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
#[cfg(test)]
mod tests {
use super::*;
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
external_user INTEGER NOT NULL,
message_time INTEGER NOT NULL,
content TEXT NOT NULL,
sent_by_self INTEGER NOT NULL,
message_state TEXT NOT NULL,
height INTEGER NOT NULL DEFAULT 0
);
#[test]
fn resumes_migration_when_height_exists_before_its_version() -> Result<(), StorageError> {
let conn = Connection::open_in_memory()?;
conn.execute_batch(
r#"
CREATE TABLE messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
external_user INTEGER NOT NULL,
message_time INTEGER NOT NULL,
content TEXT NOT NULL,
sent_by_self INTEGER NOT NULL,
message_state TEXT NOT NULL
);
ALTER TABLE messages ADD COLUMN height INTEGER NOT NULL DEFAULT 0;
PRAGMA user_version = 1;
"#,
)?;
CREATE INDEX IF NOT EXISTS idx_messages_lookup
ON messages (storage_owner, external_user, message_time DESC);
run_migrations_on_connection(&conn)?;
CREATE TABLE IF NOT EXISTS contacts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
user_id INTEGER NOT NULL,
user_name TEXT,
last_message_at INTEGER,
UNIQUE(storage_owner, user_id)
);
let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
assert_eq!(version, 5);
for column in ["height", "reply_to", "edited_count", "deleted_by_external"] {
let mut statement =
conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
assert!(statement.exists([column])?);
CREATE INDEX IF NOT EXISTS idx_contacts_owner
ON contacts (storage_owner, last_message_at DESC, user_id ASC);
CREATE TABLE IF NOT EXISTS communities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
storage_owner INTEGER NOT NULL,
address TEXT NOT NULL,
title TEXT NOT NULL,
position TEXT NOT NULL,
UNIQUE(storage_owner, address)
);
CREATE INDEX IF NOT EXISTS idx_communities_owner
ON communities (storage_owner);
"#;
match create_shared_connection("messages", INIT_SQL) {
Ok(shared_conn) => {
// Attempt to add the height column for backwards compatibility.
// This will fail if the column already exists, which is expected.
let _ = with_conn(&shared_conn, |conn| {
let _ = conn.execute(
"ALTER TABLE messages ADD COLUMN height INTEGER NOT NULL DEFAULT 0",
[],
);
Ok(())
});
// Attempt to add the reply_to column for backwards compatibility.
// This will fail if the column already exists, which is expected.
let _ = with_conn(&shared_conn, |conn| {
let _ = conn.execute(
"ALTER TABLE messages ADD COLUMN reply_to INTEGER",
[],
);
Ok(())
});
Ok(shared_conn)
}
Ok(())
Err(e) => Err(e),
}
}
/*
Example usage:
// In some util module (at init time, e.g. lazy_static or LazyLock)
static MESSAGES_DB: LazyLock<Arc<Mutex<Connection>>> = LazyLock::new(|| {
create_general_messages_db().expect("failed to create messages DB")
});
// Later, to run a query:
let res: Result<Vec<MyRow>, String> = with_conn(&MESSAGES_DB, |conn| {
let mut stmt = conn.prepare("SELECT ...")?;
let rows = stmt.query_map(...)?;
// collect and return Ok(...)
});
*/

6
iota-storage/src/util/e2ee_storage.rs
View file

@ -1,5 +1,5 @@
use crate::util::db;
use rusqlite::{OptionalExtension, params};
use rusqlite::{params, OptionalExtension};
use std::sync::{Arc, LazyLock, Mutex};
pub type StorageError = String;
@ -226,7 +226,9 @@ fn chat_secret_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<StoredChatS
})
}
fn pending_forward_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<PendingChatSecretForward> {
fn pending_forward_from_row(
row: &rusqlite::Row<'_>,
) -> rusqlite::Result<PendingChatSecretForward> {
Ok(PendingChatSecretForward {
recipient_user_id: row.get(0)?,
chat_id: row.get(1)?,

1
iota-storage/src/util/mod.rs
View file

@ -4,4 +4,3 @@ pub mod communities_util;
pub mod config_util;
pub mod db;
pub mod e2ee_storage;
pub mod settings;

163
iota-storage/src/util/settings.rs
View file

@ -1,163 +0,0 @@
use crate::storage_error::StorageError;
use crate::util::db;
use iota_util::file_util::get_directory;
use rusqlite::{OptionalExtension, params};
use std::fs;
use std::path::Path;
pub const GLOBAL_SESSION_ID: i64 = 0;
const GLOBAL_SETTINGS_NAME: &str = "__global__";
pub fn save(user_id: i64, session_id: i64, name: &str, payload: &str) -> Result<(), StorageError> {
db::with_db(|conn| {
conn.execute(
"INSERT INTO settings (user_id, session_id, name, payload) VALUES (?1, ?2, ?3, ?4)\n ON CONFLICT(user_id, session_id, name) DO UPDATE SET payload = excluded.payload",
params![user_id, session_id, name, payload],
)?;
Ok(())
})
}
pub fn load(user_id: i64, session_id: i64, name: &str) -> Result<Option<String>, StorageError> {
db::with_db(|conn| {
conn.query_row(
"SELECT payload FROM settings WHERE user_id = ?1 AND session_id = ?2 AND name = ?3",
params![user_id, session_id, name],
|row| row.get(0),
)
.optional()
.map_err(StorageError::from)
})
}
pub fn list(user_id: i64, session_id: i64) -> Result<Vec<String>, StorageError> {
db::with_db(|conn| {
let mut statement = conn.prepare(
"SELECT name FROM settings WHERE user_id = ?1 AND session_id = ?2 ORDER BY name",
)?;
let rows = statement.query_map(params![user_id, session_id], |row| row.get(0))?;
rows.collect::<Result<Vec<String>, _>>()
.map_err(StorageError::from)
})
}
pub fn save_global(user_id: i64, payload: &str) -> Result<(), StorageError> {
save(user_id, GLOBAL_SESSION_ID, GLOBAL_SETTINGS_NAME, payload)
}
pub fn load_global(user_id: i64) -> Result<Option<String>, StorageError> {
load(user_id, GLOBAL_SESSION_ID, GLOBAL_SETTINGS_NAME)
}
pub fn migrate_legacy_files() -> Result<(), StorageError> {
let users_dir = Path::new(&get_directory()).join("users");
let Ok(users) = fs::read_dir(users_dir) else {
return Ok(());
};
for user_entry in users {
let user_entry = user_entry?;
let Ok(user_id) = user_entry.file_name().to_string_lossy().parse::<i64>() else {
continue;
};
let user_dir = user_entry.path();
migrate_file_if_missing(
user_id,
GLOBAL_SESSION_ID,
GLOBAL_SETTINGS_NAME,
&user_dir.join("global.settings"),
)?;
let settings_dir = user_dir.join("settings");
let Ok(settings_entries) = fs::read_dir(settings_dir) else {
continue;
};
for settings_entry in settings_entries {
let settings_entry = settings_entry?;
let path = settings_entry.path();
if path.is_file() {
if let Some(name) = setting_name(&path) {
migrate_file_if_missing(user_id, GLOBAL_SESSION_ID, &name, &path)?;
}
continue;
}
let Ok(session_id) = settings_entry.file_name().to_string_lossy().parse::<i64>() else {
continue;
};
let Ok(device_settings) = fs::read_dir(path) else {
continue;
};
for setting_entry in device_settings {
let setting_entry = setting_entry?;
let path = setting_entry.path();
if let Some(name) = setting_name(&path) {
migrate_file_if_missing(user_id, session_id, &name, &path)?;
}
}
}
}
Ok(())
}
fn setting_name(path: &Path) -> Option<String> {
(path.extension()?.to_str()? == "settings").then(|| {
path.file_stem()
.and_then(|name| name.to_str())
.unwrap_or_default()
.to_string()
})
}
fn migrate_file_if_missing(
user_id: i64,
session_id: i64,
name: &str,
path: &Path,
) -> Result<(), StorageError> {
if !path.is_file() || load(user_id, session_id, name)?.is_some() {
return Ok(());
}
let payload = fs::read_to_string(path)?;
save(user_id, session_id, name, &payload)
}
#[cfg(test)]
mod tests {
use super::*;
use rusqlite::Connection;
#[test]
fn settings_schema_supports_user_and_session_keys() -> Result<(), StorageError> {
let conn = Connection::open_in_memory()?;
conn.execute_batch(
"CREATE TABLE settings (
user_id INTEGER NOT NULL,
session_id INTEGER NOT NULL,
name TEXT NOT NULL,
payload TEXT NOT NULL,
PRIMARY KEY (user_id, session_id, name)
);",
)?;
conn.execute(
"INSERT INTO settings VALUES (?1, ?2, ?3, ?4)",
params![7, 11, "theme", "dark"],
)?;
conn.execute(
"INSERT INTO settings VALUES (?1, ?2, ?3, ?4)",
params![7, 12, "theme", "light"],
)?;
let payload: String = conn.query_row(
"SELECT payload FROM settings WHERE user_id = 7 AND session_id = 11 AND name = 'theme'",
[],
|row| row.get(0),
)?;
assert_eq!(payload, "dark");
Ok(())
}
}

5
iota-util/Cargo.toml
View file

@ -4,9 +4,8 @@ version = "0.1.0"
edition = "2024"
[dependencies]
mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = [
"crypto"
] }
mtp = { git = "https://git.methanium.net/Methanium/mtp.git" }
mtp-crypto = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["pqc"] }
reqwest = "0.13.2"
tokio = { version = "1.50.0", features = ["full"] }

4
iota-util/src/crypto_helper.rs
View file

@ -1,5 +1,5 @@
use base64::{Engine as _, engine::general_purpose::STANDARD};
use mtp::crypto::{Keyring, PublicKeyBundle};
use mtp_crypto::{Keyring, PublicKeyBundle};
pub fn generate_keyring() -> Keyring {
Keyring::generate()
@ -24,5 +24,5 @@ pub fn public_key_bundle_from_base64(s: &str) -> Option<PublicKeyBundle> {
}
pub fn hex_hash(input: &str) -> String {
hex::encode(mtp::crypto::sha256(input.as_bytes()))
hex::encode(mtp_crypto::sha256(input.as_bytes()))
}

21
iota-util/src/crypto_util.rs
View file

@ -1,5 +1,5 @@
use base64::{Engine as _, engine::general_purpose::STANDARD};
use mtp::crypto::{EncryptionType, Keyring, PublicKeyBundle, decrypt_with, encrypt_for};
use mtp_crypto::{EncryptionType, Keyring, PublicKeyBundle, encrypt_for, decrypt_with};
#[derive(Clone, Copy, Debug)]
pub enum DataFormat {
@ -22,8 +22,13 @@ pub fn encrypt(
.map_err(|e| format!("encryption error: {:?}", e))
}
pub fn decrypt(ciphertext: &[u8], aad: &[u8], keyring: &Keyring) -> Result<Vec<u8>, String> {
decrypt_with(ciphertext, keyring, aad).map_err(|e| format!("decryption error: {:?}", e))
pub fn decrypt(
ciphertext: &[u8],
aad: &[u8],
keyring: &Keyring,
) -> Result<Vec<u8>, String> {
decrypt_with(ciphertext, keyring, aad)
.map_err(|e| format!("decryption error: {:?}", e))
}
pub fn encrypt_challenge(
@ -34,10 +39,12 @@ pub fn encrypt_challenge(
Ok(STANDARD.encode(&blob))
}
pub fn decrypt_challenge(encrypted: &str, keyring: &Keyring) -> Result<String, String> {
let blob = STANDARD
.decode(encrypted)
.map_err(|e| format!("base64 decode error: {}", e))?;
pub fn decrypt_challenge(
encrypted: &str,
keyring: &Keyring,
) -> Result<String, String> {
let blob =
STANDARD.decode(encrypted).map_err(|e| format!("base64 decode error: {}", e))?;
let pt = decrypt(&blob, b"challenge", keyring)?;
String::from_utf8(pt).map_err(|e| format!("utf8 decode error: {}", e))
}

1
omikron-connector/Cargo.toml
View file

@ -4,7 +4,6 @@ version = "0.1.0"
edition = "2024"
[dependencies]
iota-connection = { path = "../iota-connection" }
iota-logger = { path = "../iota-logger" }
iota-state = { path = "../iota-state" }
iota-storage = { path = "../iota-storage" }

2494
omikron-connector/src/omikron_connection.rs
View file

File diff suppressed because it is too large Load diff

7
omikron-connector/src/ping_pong_task.rs
View file

@ -2,7 +2,6 @@ use crate::omikron_connection::OmikronConnection;
use dashmap::DashMap;
use iota_state::APP_STATE;
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
use std::sync::atomic::Ordering;
use std::sync::LazyLock;
use std::time::Instant;
use tokio::time::Duration;
@ -17,8 +16,6 @@ impl OmikronConnection {
PING_TIMES.retain(|_, v| v.elapsed() < Duration::from_secs(30));
self.missed_pongs.fetch_add(1, Ordering::Relaxed);
let ping_message = CommunicationValue::new(CommunicationType::Ping)
.with_id(id)
.add_typed_default(
@ -26,12 +23,10 @@ impl OmikronConnection {
DataValue::Array(vec![DataValue::SignedNumber(*self.last_ping.lock().await as i128)]),
);
let _ = self.send_message(&ping_message).await;
self.send_message(&ping_message).await;
}
pub async fn handle_pong(&self, cv: &CommunicationValue) {
self.missed_pongs.store(0, Ordering::Relaxed);
let id = cv.get_id();
if let Some((_, send_time)) = PING_TIMES.remove(&id) {

3
type-maps.yaml
View file

@ -148,7 +148,6 @@ type_maps:
MessageReactionAdd: 146
MessageReactionRemove: 147
MessageReactionLive: 148
MessageDeleteLive: 150
DataTypes:
ErrorType: 32
ErrorProtocol: 33
@ -170,7 +169,7 @@ type_maps:
CallState: 49
ScreenShare: 50
PrivateKeyHash: 51
# Accepted: 52 now part of default MTP
Accepted: 52
AcceptedProfiles: 53
DeniedProfiles: 54
Content: 55

8
web-server/Cargo.toml
View file

@ -5,10 +5,4 @@ version = "0.1.0"
edition = "2024"
[dependencies]
mtp = { git = "https://git.methanium.net/Methanium/mtp.git", features = ["web-server"] }
bytes = "1"
http = "1"
iota-state = { path = "../iota-state" }
iota-util = { path = "../iota-util" }
iota-logger = { path = "../iota-logger" }
tokio = { version = "1.50.0", features = ["full"] }
mtp = { git = "https://git.methanium.net/Methanium/mtp.git" }

139
web-server/src/lib.rs
View file

@ -1,133 +1,6 @@
use bytes::Bytes;
use iota_logger::log;
use iota_state::{ACTIVE_TASKS, SHUTDOWN};
use iota_util::file_util::load_file_vec;
use mtp::host::HostConfig;
use mtp::webserver::{Http3Request, Http3Response, MTPWebServer, WebServerConfig};
use std::net::{IpAddr, Ipv4Addr};
use tokio::time::{Duration, sleep};
const CERT_PATH: &str = "certs/cert.pem";
const KEY_PATH: &str = "certs/cert.key";
async fn root(_request: Http3Request, response: Http3Response) -> Http3Response {
static_file("index.html", response).await
}
async fn static_file(path: &str, response: Http3Response) -> Http3Response {
let file = path.trim_start_matches('/');
let file = if file.is_empty() { "index.html" } else { file };
if file.split('/').any(|component| component == "..") {
return response
.status(http::StatusCode::BAD_REQUEST)
.body("invalid path");
}
let path = std::path::Path::new("web").join(file);
let Some(parent) = path.parent().and_then(|path| path.to_str()) else {
return response
.status(http::StatusCode::NOT_FOUND)
.body("not found");
};
let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
return response
.status(http::StatusCode::NOT_FOUND)
.body("not found");
};
match load_file_vec(parent, name) {
Ok(body) => response
.status(http::StatusCode::OK)
.header("content-type", content_type(name))
.body(Bytes::from(body)),
Err(_) => response
.status(http::StatusCode::NOT_FOUND)
.body("not found"),
}
}
fn content_type(name: &str) -> &'static str {
match std::path::Path::new(name)
.extension()
.and_then(|ext| ext.to_str())
{
Some("html") => "text/html; charset=utf-8",
Some("css") => "text/css; charset=utf-8",
Some("js") => "application/javascript; charset=utf-8",
Some("json") => "application/json",
Some("png") => "image/png",
Some("ico") => "image/x-icon",
Some("woff2") => "font/woff2",
_ => "application/octet-stream",
}
}
pub async fn start(port: u16) -> bool {
let certificate = match tokio::fs::read(CERT_PATH).await {
Ok(certificate) => certificate,
Err(error) => {
log!("MTP web server certificate load failed: {}", error);
return false;
}
};
let key = match tokio::fs::read(KEY_PATH).await {
Ok(key) => key,
Err(error) => {
log!("MTP web server key load failed: {}", error);
return false;
}
};
let host_config = HostConfig::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port, certificate, key);
let web_config = match WebServerConfig::new().route("/", root).and_then(|config| {
config.fallback(|request, response| async move {
static_file(request.uri.path(), response).await
})
}) {
Ok(config) => config,
Err(error) => {
log!("MTP web server route setup failed: {}", error);
return false;
}
};
let mut server = match MTPWebServer::new(host_config, web_config).await {
Ok(server) => server,
Err(error) => {
log!("MTP web server startup failed: {}", error);
return false;
}
};
log!("MTP web server running on port {}", port);
tokio::spawn(async move {
ACTIVE_TASKS.insert("WebServer".into());
loop {
tokio::select! {
result = server.accept() => {
match result {
Ok(Some(_connection)) => {}
Ok(None) => break,
Err(error) => log!("MTP webserver connection failed: {}", error),
}
}
_ = wait_for_shutdown() => {
server.shutdown().await;
break;
}
}
}
ACTIVE_TASKS.remove("WebServer");
});
true
}
async fn wait_for_shutdown() {
loop {
if *SHUTDOWN.read().await {
break;
}
sleep(Duration::from_millis(100)).await;
}
}
// The web server is a TTP host & identification system,
// it "upgrades" connections after identification to
//
// either Own User (Cut down version of the Omikron Connection),
// or Community (Custom Connection),
// or Iota (Custom Connection).