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

Merge branch 'main' of ssh://github.com/Tensamin/Omega

Browse source
This commit is contained in:
Alex Emmet 2026-03-03 10:28:32 +01:00
commit e35ae230ee
13 changed files with 754 additions and 1617 deletions
Download patch file Download diff file Expand all files Collapse all files

403
src/data/communication.rs
View file

@ -1,403 +0,0 @@
use json::number::Number;
use json::{Array, JsonValue, object, parse};
use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
use strum::IntoEnumIterator;
use strum_macros::EnumIter;
use uuid::Uuid;
#[derive(Eq, Hash, PartialEq, EnumIter, Clone, Debug)]
#[allow(non_camel_case_types, dead_code)]
pub enum DataTypes {
error_type,
accepted_ids,
uuid,
register_id,
link,
settings,
settings_name,
chat_partner_id,
chat_partner_name,
iota_id,
user_id,
user_ids,
iota_ids,
user_state,
user_states,
user_pings,
call_state,
screen_share,
private_key_hash,
accepted,
accepted_profiles,
denied_profiles,
content,
messages,
notifications,
send_time,
get_time,
get_variant,
shared_secret_own,
shared_secret_other,
shared_secret_sign,
shared_secret,
call_id,
call_token,
untill,
enabled,
start_date,
end_date,
receiver_id,
sender_id,
signature,
signed,
message,
message_state,
last_ping,
ping_iota,
ping_clients,
matches,
omikron,
offset,
amount,
position,
name,
path,
codec,
function,
payload,
result,
interactables,
want_to_watch,
watcher,
created_at,
username,
display,
avatar,
about,
status,
public_key,
sub_level,
sub_end,
community_address,
challenge,
community_title,
communities,
rho_connections,
user,
online_status,
omikron_id,
omikron_connections,
reset_token,
new_token,
}
impl DataTypes {
pub fn parse(p0: String) -> DataTypes {
for datatype in DataTypes::iter() {
if datatype.to_string().to_lowercase().replace('_', "")
== p0.to_lowercase().replace('_', "")
{
return datatype;
}
}
DataTypes::error_type
}
pub fn to_string(&self) -> String {
return format!("{:?}", self);
}
}
#[derive(PartialEq, Clone, EnumIter, Debug)]
#[allow(non_camel_case_types, dead_code)]
pub enum CommunicationType {
error,
error_anonymous,
error_internal,
error_invalid_data,
error_invalid_user_id,
error_invalid_omikron_id,
error_not_found,
error_not_authenticated,
error_no_iota,
error_invalid_challenge,
error_invalid_secret,
error_invalid_private_key,
error_invalid_public_key,
error_no_user_id,
error_no_call_id,
error_invalid_call_id,
success,
shorten_link,
settings_save,
settings_load,
settings_list,
message,
message_state,
message_send,
message_live,
message_other_iota,
message_chunk,
messages_get,
push_notification,
read_notification,
get_notifications,
change_confirm,
confirm_receive,
confirm_read,
get_chats,
get_states,
add_community,
remove_community,
get_communities,
challenge,
challenge_response,
register,
register_response,
identification,
identification_response,
register_iota,
register_iota_success,
ping,
pong,
add_conversation,
send_chat,
client_changed,
client_connected,
client_disconnected,
client_closed,
public_key,
private_key,
webrtc_sdp,
webrtc_ice,
start_stream,
end_stream,
watch_stream,
call_token,
call_invite,
call_disconnect_user,
call_timeout_user,
call_set_anonymous_joining,
end_call,
function,
update,
create_user,
rho_update,
user_connected,
user_disconnected,
iota_connected,
iota_disconnected,
sync_client_iota_status,
get_user_data,
get_iota_data,
iota_user_data,
change_user_data,
change_iota_data,
get_register,
complete_register_user,
complete_register_iota,
delete_user,
delete_iota,
start_register,
complete_register,
}
impl CommunicationType {
pub fn parse(p0: String) -> CommunicationType {
for datatype in CommunicationType::iter() {
if datatype.to_string().to_lowercase().replace('_', "")
== p0.to_lowercase().replace('_', "")
{
return datatype;
}
}
CommunicationType::error
}
pub fn to_string(&self) -> String {
return format!("{:?}", self);
}
}
#[derive(Debug, Clone)]
pub struct CommunicationValue {
id: Uuid,
comm_type: CommunicationType,
sender: i64,
receiver: i64,
data: HashMap<DataTypes, JsonValue>,
}
#[allow(dead_code)]
impl CommunicationValue {
pub fn new(comm_type: CommunicationType) -> Self {
Self {
id: Uuid::new_v4(),
comm_type,
sender: 0,
receiver: 0,
data: HashMap::new(),
}
}
pub fn with_id(mut self, p0: Uuid) -> Self {
self.id = p0;
self
}
pub fn get_id(&self) -> Uuid {
self.id.clone()
}
pub fn with_sender(mut self, sender: i64) -> Self {
self.sender = sender;
self
}
pub fn get_sender(&self) -> i64 {
self.sender.clone()
}
pub fn with_receiver(mut self, receiver: i64) -> Self {
self.receiver = receiver;
self
}
pub fn get_receiver(&self) -> i64 {
self.receiver.clone()
}
pub fn add_data_num(mut self, key: DataTypes, value: Number) -> Self {
self.data.insert(key, JsonValue::Number(value));
self
}
pub fn add_data_str(mut self, key: DataTypes, value: String) -> Self {
self.data.insert(key, JsonValue::String(value));
self
}
pub fn add_data(mut self, key: DataTypes, value: JsonValue) -> Self {
self.data.insert(key, value);
self
}
pub fn add_array(mut self, key: DataTypes, value: Array) -> Self {
self.data.insert(key, JsonValue::Array(value));
self
}
pub fn get_data(&self, key: DataTypes) -> Option<&JsonValue> {
self.data.get(&key)
}
pub fn get_type(&self) -> CommunicationType {
self.comm_type.clone()
}
pub fn is_type(&self, p0: CommunicationType) -> bool {
self.comm_type == p0
}
pub fn to_json(&self) -> JsonValue {
let mut jdata = object! {};
for (k, v) in &self.data {
jdata[&format!("{:?}", k)] = JsonValue::from(v.clone());
}
if self.sender > 0 && self.receiver > 0 {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
sender: self.sender,
receiver: self.receiver,
data: jdata
}
} else if self.sender > 0 {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
sender: self.sender,
data: jdata
}
} else if self.receiver > 0 {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
receiver: self.receiver,
data: jdata
}
} else {
object! {
id: self.id.to_string(),
type: format!("{:?}", self.comm_type),
data: jdata
}
}
}
pub fn from_json(json_str: &str) -> Self {
if let Ok(parsed) = parse(json_str) {
let comm_type = CommunicationType::parse(parsed["type"].to_string());
let mut sender: i64 = 0;
if parsed.has_key("sender") {
sender = parsed["sender"].as_i64().unwrap_or(0);
}
let mut receiver: i64 = 0;
if parsed.has_key("receiver") {
receiver = parsed["receiver"].as_i64().unwrap_or(0);
}
let uuid =
Uuid::parse_str(parsed["id"].as_str().unwrap_or("")).unwrap_or(Uuid::new_v4());
let mut data = HashMap::new();
if parsed["data"].is_object() {
for (k, v) in parsed["data"].entries() {
data.insert(DataTypes::parse(k.to_string()), v.clone());
}
}
Self {
id: uuid,
comm_type,
sender,
receiver,
data,
}
} else {
Self {
id: Uuid::new_v4(),
comm_type: CommunicationType::error,
sender: 0,
receiver: 0,
data: HashMap::new(),
}
}
}
pub fn forward_to_other_iota(original: &mut CommunicationValue) -> CommunicationValue {
let receiver = original
.get_data(DataTypes::receiver_id)
.unwrap_or(&JsonValue::Number(Number::from(0)))
.as_i64()
.unwrap_or(0);
let now_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_millis() as i64;
let sender = original.get_sender();
CommunicationValue::new(CommunicationType::message_other_iota)
.with_id(original.get_id())
.with_receiver(receiver)
.add_data(
DataTypes::receiver_id,
JsonValue::Number(Number::from(receiver)),
)
.with_sender(sender)
.add_data(DataTypes::send_time, JsonValue::String(now_ms.to_string()))
.add_data(
DataTypes::sender_id,
JsonValue::Number(Number::from(sender)),
)
.add_data(
DataTypes::content,
JsonValue::String(original.get_data(DataTypes::content).unwrap().to_string()),
)
}
}

1
src/data/mod.rs
View file

@ -1 +0,0 @@
pub mod communication;

6
src/main.rs
View file

@ -1,8 +1,8 @@
mod data;
mod server;
mod sql;
mod util;
use crate::server::omikron_connection;
use crate::sql::sql::initialize_db;
use crate::sql::sql::print_users;
use crate::util::crypto_helper::load_public_key;
@ -43,7 +43,7 @@ async fn main() {
} else {
log!(" Users");
}
let _ = server::server::start(9187).await;
let _ = server::server::start(9188).await;
let _ = omikron_connection::OmikronServer::start(9187).await;
tokio::signal::ctrl_c().await.unwrap();
}

234
src/server/api.rs
View file

@ -1,4 +1,3 @@
use crate::data::communication::{CommunicationType, CommunicationValue, DataTypes};
use crate::get_public_key;
use crate::server::omikron_manager::get_random_omikron;
use crate::sql::sql;
@ -11,7 +10,6 @@ use actix_web::HttpResponse;
use actix_web::http::{StatusCode, header};
use base64::Engine as _;
use json::JsonValue;
use json::number::Number;
pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
if path == "OPTIONS" {
@ -35,6 +33,9 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
};
let (status, body_text) = match path_parts.as_slice() {
// ==================================================
// DOWNLOAD IOTA FRONTEND
// ==================================================
["api", "download", "iota_frontend"] => {
let file_path = "downloads/[iota_frontend].zip";
@ -50,87 +51,107 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
.body(file_bytes);
}
Err(_) => {
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
return HttpResponse::NotFound()
.insert_header(("Access-Control-Allow-Origin", "*"))
.body("File not found");
.body(res.dump());
}
}
}
// ==================================================
// GET RANDOM OMIKRON
// ==================================================
["api", "get", "omikron"] => {
if let Ok(omikron_conn) = get_random_omikron().await {
if let Ok((public_key, ip_address)) =
sql::get_omikron_by_id(omikron_conn.get_omikron_id().await).await
{
(
StatusCode::OK,
format!(
"{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
omikron_conn.get_omikron_id().await,
public_key,
ip_address
),
)
let id = omikron_conn.get_omikron_id().await;
if let Ok((public_key, ip_address)) = sql::get_omikron_by_id(id).await {
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = id.into();
res["public_key"] = public_key.into();
res["ip_address"] = ip_address.into();
(StatusCode::OK, res.dump())
} else {
(
StatusCode::INTERNAL_SERVER_ERROR,
"selected an invalid omikron".to_string(),
)
let mut res = JsonValue::new_object();
res["status"] = "error".into();
(StatusCode::INTERNAL_SERVER_ERROR, res.dump())
}
} else {
(
StatusCode::NOT_FOUND,
"couldn't find online omikron".to_string(),
)
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::NOT_FOUND, res.dump())
}
}
// ==================================================
// GET OMIKRON BY ID
// ==================================================
["api", "get", "omikron", id] => {
let id = id.parse::<i64>().unwrap_or(0);
if id == 0 {
not_found()
let mut res = JsonValue::new_object();
res["status"] = "error_bad_request".into();
(StatusCode::BAD_REQUEST, res.dump())
} else if let Ok((public_key, ip_address)) = get_omikron_by_id(id).await {
(
StatusCode::OK,
format!(
"{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
id, public_key, ip_address
),
)
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = id.into();
res["public_key"] = public_key.into();
res["ip_address"] = ip_address.into();
(StatusCode::OK, res.dump())
} else if let Some(omikron_id) = get_iota_primary_omikron_connection(id) {
if let Ok((public_key, ip_address)) = get_omikron_by_id(omikron_id).await {
(
StatusCode::OK,
format!(
"{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
omikron_id, public_key, ip_address
),
)
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = omikron_id.into();
res["public_key"] = public_key.into();
res["ip_address"] = ip_address.into();
(StatusCode::OK, res.dump())
} else {
not_found()
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::NOT_FOUND, res.dump())
}
} else if let Ok((_, iota_id, _, _, _, _, _, _, _, _, _, _)) = get_by_user_id(id).await
{
if let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) {
if let Ok((public_key, ip_address)) = get_omikron_by_id(omikron_id).await {
(
StatusCode::OK,
format!(
"{{\"id\": {}, \"public_key\": \"{}\", \"ip_address\": \"{}\"}}",
omikron_id, public_key, ip_address
),
)
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = omikron_id.into();
res["public_key"] = public_key.into();
res["ip_address"] = ip_address.into();
(StatusCode::OK, res.dump())
} else {
not_found()
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::NOT_FOUND, res.dump())
}
} else {
not_found()
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::NOT_FOUND, res.dump())
}
} else {
not_found()
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::NOT_FOUND, res.dump())
}
}
// ==================================================
// GET ID BY USERNAME
// ==================================================
["api", "get", "id", username] => {
if username.is_empty() {
not_found()
let mut res = JsonValue::new_object();
res["status"] = "error_bad_request".into();
(StatusCode::BAD_REQUEST, res.dump())
} else if let Ok((
id,
iota_id,
@ -146,37 +167,49 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
_,
)) = sql::get_by_username(username).await
{
let cv = CommunicationValue::new(CommunicationType::success)
.add_data_str(DataTypes::username, username)
.add_data_str(DataTypes::public_key, public_key)
.add_data(DataTypes::user_id, JsonValue::Number(Number::from(id)))
.add_data(DataTypes::iota_id, JsonValue::Number(Number::from(iota_id)))
.add_data(
DataTypes::sub_level,
JsonValue::Number(Number::from(sub_level)),
)
.add_data(DataTypes::sub_end, JsonValue::Number(Number::from(sub_end)));
(StatusCode::OK, cv.to_json().to_string())
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["username"] = username.into();
res["public_key"] = public_key.into();
res["user_id"] = id.into();
res["iota_id"] = iota_id.into();
res["sub_level"] = sub_level.into();
res["sub_end"] = sub_end.into();
(StatusCode::OK, res.dump())
} else {
(
StatusCode::OK,
CommunicationValue::new(CommunicationType::error_not_found)
.to_json()
.to_string(),
)
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::OK, res.dump())
}
}
["api", "get", "public_key"] => (StatusCode::OK, public_key_to_base64(&get_public_key())),
// ==================================================
// GET SERVER PUBLIC KEY
// ==================================================
["api", "get", "public_key"] => {
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["public_key"] = public_key_to_base64(&get_public_key()).into();
(StatusCode::OK, res.dump())
}
// ==================================================
// GET USER BY ID
// ==================================================
["api", "get", "user", id] => {
let id: i64 = id.parse().unwrap_or(0);
if id == 0 {
bad_request()
let mut res = JsonValue::new_object();
res["status"] = "error_bad_request".into();
(StatusCode::BAD_REQUEST, res.dump())
} else if let Ok((
id,
iota_id,
username,
display,
status,
status_msg,
about,
avatar,
sub_level,
@ -186,47 +219,46 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
_,
)) = sql::get_by_user_id(id).await
{
let mut cv = CommunicationValue::new(CommunicationType::success)
.add_data_str(DataTypes::username, username)
.add_data_str(DataTypes::public_key, public_key)
.add_data(DataTypes::user_id, JsonValue::Number(Number::from(id)))
.add_data(DataTypes::iota_id, JsonValue::Number(Number::from(iota_id)))
.add_data(
DataTypes::sub_level,
JsonValue::Number(Number::from(sub_level)),
)
.add_data(DataTypes::sub_end, JsonValue::Number(Number::from(sub_end)));
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["username"] = username.into();
res["public_key"] = public_key.into();
res["user_id"] = id.into();
res["iota_id"] = iota_id.into();
res["sub_level"] = sub_level.into();
res["sub_end"] = sub_end.into();
if let Some(display) = display {
cv = cv.add_data_str(DataTypes::display, display);
res["display"] = display.into();
}
if let Some(status) = status {
cv = cv.add_data_str(DataTypes::status, status);
if let Some(status_msg) = status_msg {
res["status_message"] = status_msg.into();
}
if let Some(about) = about {
cv = cv.add_data_str(DataTypes::about, about);
res["about"] = about.into();
}
if let Some(avatar) = avatar {
cv = cv.add_data_str(
DataTypes::avatar,
base64::engine::general_purpose::STANDARD.encode(avatar),
);
res["avatar"] = base64::engine::general_purpose::STANDARD
.encode(avatar)
.into();
}
(StatusCode::OK, cv.to_json().to_string())
(StatusCode::OK, res.dump())
} else {
(
StatusCode::OK,
CommunicationValue::new(CommunicationType::error_not_found)
.to_json()
.to_string(),
)
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::OK, res.dump())
}
}
_ => (
StatusCode::INTERNAL_SERVER_ERROR,
CommunicationValue::new(CommunicationType::error)
.to_json()
.to_string(),
),
// ==================================================
// DEFAULT
// ==================================================
_ => {
let mut res = JsonValue::new_object();
res["status"] = "error".into();
(StatusCode::INTERNAL_SERVER_ERROR, res.dump())
}
};
let body_bytes = body_text.into_bytes();

1
src/server/mod.rs
View file

@ -3,4 +3,3 @@ pub mod omikron_connection;
pub mod omikron_manager;
pub mod server;
pub mod short_link;
pub mod socket;

1093
src/server/omikron_connection.rs
View file

File diff suppressed because it is too large Load diff

14
src/server/server.rs
View file

@ -1,11 +1,10 @@
use crate::{
log,
server::{api, short_link::get_short_link, socket},
server::{api, short_link::get_short_link},
util::file_util::get_directory,
};
use actix_web::{App, HttpRequest, HttpResponse, HttpServer, Responder, http::header, web};
use actix_web_actors::ws;
use rustls::ServerConfig;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
@ -42,7 +41,6 @@ pub async fn start(port: u16) -> anyhow::Result<()> {
App::new()
.route("/api/{path:.*}", web::to(api_handler))
.route("/direct/{path:.*}", web::to(direct_handler))
.route("/ws/{path:.*}", web::get().to(ws_handler))
})
.bind_rustls_0_23(addr, config)?
.run()
@ -64,16 +62,6 @@ async fn direct_handler(req: HttpRequest) -> impl Responder {
.finish()
}
}
async fn ws_handler(
req: HttpRequest,
stream: web::Payload,
path: web::Path<String>,
) -> Result<HttpResponse, actix_web::Error> {
let path = path.into_inner();
println!("WS handler reached: {}", path);
ws::start(socket::WsSession::new(path), &req, stream)
}
async fn api_handler(req: HttpRequest, body: web::Bytes) -> HttpResponse {
let path = req.uri().path().to_string();

129
src/server/socket.rs
View file

@ -1,129 +0,0 @@
use std::sync::Arc;
use std::time::{Duration, Instant};
use actix::{Actor, ActorContext, AsyncContext, StreamHandler};
use actix_web_actors::ws;
use crate::data::communication::{CommunicationType, CommunicationValue};
use crate::log;
use crate::server::omikron_connection::OmikronConnection;
const IDLE_TIMEOUT: Duration = Duration::from_secs(30);
use actix::Message;
#[derive(Message)]
#[rtype(result = "()")]
pub struct WsSendMessage(pub String);
impl actix::Handler<WsSendMessage> for WsSession {
type Result = ();
fn handle(&mut self, msg: WsSendMessage, ctx: &mut Self::Context) {
ctx.text(msg.0);
}
}
pub struct WsSession {
path: String,
last_heartbeat: Instant,
omikron: Option<Arc<OmikronConnection>>,
}
impl WsSession {
pub fn new(path: String) -> Self {
Self {
path,
last_heartbeat: Instant::now(),
omikron: None,
}
}
fn start_heartbeat(&self, ctx: &mut ws::WebsocketContext<Self>) {
ctx.run_interval(Duration::from_secs(5), |act, ctx| {
if Instant::now().duration_since(act.last_heartbeat) > IDLE_TIMEOUT {
log!("[ws_handler] Heartbeat failed. Disconnecting.");
ctx.close(None);
ctx.stop();
return;
}
let ping = CommunicationValue::new(CommunicationType::ping)
.to_json()
.to_string();
ctx.text(ping);
});
}
}
impl Actor for WsSession {
type Context = ws::WebsocketContext<Self>;
fn started(&mut self, ctx: &mut Self::Context) {
self.start_heartbeat(ctx);
if self.path == "omikron" {
let addr = ctx.address();
let connection = OmikronConnection::new(addr);
self.omikron = Some(connection);
}
}
fn stopped(&mut self, _: &mut Self::Context) {
log!(
"[ws] WebSocket handling task for path: {} is finished.",
self.path
);
if let Some(conn) = &self.omikron {
let conn = conn.clone();
actix_rt::spawn(async move {
conn.handle_close().await;
});
}
}
}
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
match msg {
Ok(ws::Message::Text(text)) => {
self.last_heartbeat = Instant::now();
if let Some(conn) = &self.omikron {
let conn_clone = conn.clone();
actix_rt::spawn(async move {
conn_clone.handle_message(text.to_string()).await;
});
}
}
Ok(ws::Message::Ping(msg)) => {
self.last_heartbeat = Instant::now();
ctx.pong(&msg);
}
Ok(ws::Message::Pong(_)) => {
self.last_heartbeat = Instant::now();
log!("[ws_handler] Received Pong. Connection alive.");
}
Ok(ws::Message::Close(reason)) => {
log!("[ws_handler] Received Close. Disconnecting.");
ctx.close(reason);
ctx.stop();
}
Ok(ws::Message::Binary(_)) => {
log!("[ws_handler] Binary message ignored.");
}
Err(e) => {
log!("[ERROR] WS Error: {}. Closing.", e);
ctx.stop();
}
_ => {}
}
}
}

4
src/sql/sql.rs
View file

@ -65,9 +65,9 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
about VARCHAR(200) COLLATE utf8mb4_bin,
avatar MEDIUMBLOB,
sub_level INT(11) NOT NULL DEFAULT 0,
sub_end BIGINT(20) NOT NULL,
sub_end BIGINT(20) NOT NULL DEFAULT 0,
public_key TEXT NOT NULL COLLATE utf8mb4_bin,
private_key_hash TEXT NOT NULL COLLATE utf8mb4_bin,
private_key_hash TEXT NOT NULL COLLATE utf8mb4_bin DEFAULT '',
iota_id BIGINT UNSIGNED NOT NULL,
token VARCHAR(255) NOT NULL UNIQUE COLLATE utf8mb4_bin
)",

65
src/sql/user_online_tracker.rs
View file

@ -33,23 +33,30 @@ pub fn track_iota_connection(iota_id: i64, omikron_id: i64, primary: bool) {
}
pub fn untrack_iota_connection(iota_id: i64, omikron_id: i64) -> bool {
if let Some(mut connections) = IOTA_OMIKRON_CONNECTIONS.get_mut(&iota_id) {
connections.retain(|&id| id != omikron_id);
let connections_empty = if let Some(r) = IOTA_OMIKRON_CONNECTIONS.get(&iota_id) {
let mut vec = r.value().clone();
vec.retain(|&id| id != omikron_id);
let empty = vec.is_empty();
drop(r);
IOTA_OMIKRON_CONNECTIONS.insert(iota_id, vec);
empty
} else {
false
};
if IOTA_PRIMARY_OMIKRON_CONNECTION
.get(&iota_id)
.map(|p| *p == omikron_id)
.unwrap_or(false)
{
if let Some(primary_ref) = IOTA_PRIMARY_OMIKRON_CONNECTION.get(&iota_id) {
let primary_id = *primary_ref.value();
drop(primary_ref);
if primary_id == omikron_id {
IOTA_PRIMARY_OMIKRON_CONNECTION.remove(&iota_id);
}
if connections.is_empty() {
IOTA_OMIKRON_CONNECTIONS.remove(&iota_id);
return true;
}
}
false
if connections_empty {
IOTA_OMIKRON_CONNECTIONS.remove(&iota_id);
}
connections_empty
}
pub fn get_iota_primary_omikron_connection(iota_id: i64) -> Option<i64> {
@ -92,28 +99,46 @@ pub async fn untrack_omikron(omikron_id: i64) {
}
let mut offline_iotas = Vec::new();
let mut primary_to_remove = Vec::new();
for mut entry in IOTA_OMIKRON_CONNECTIONS.iter_mut() {
let connections = entry.value_mut();
connections.retain(|id| *id != omikron_id);
// Collect iotas and primary info first
for r in IOTA_OMIKRON_CONNECTIONS.iter() {
let iota_id = *r.key();
let mut connections = r.value().clone();
connections.retain(|&id| id != omikron_id);
if connections.is_empty() {
offline_iotas.push(*entry.key());
offline_iotas.push(iota_id);
}
if IOTA_PRIMARY_OMIKRON_CONNECTION
.get(&iota_id)
.map(|p| *p == omikron_id)
.unwrap_or(false)
{
primary_to_remove.push(iota_id);
}
// Update the connections vector after filtering
IOTA_OMIKRON_CONNECTIONS.insert(iota_id, connections);
}
for iota_id in &offline_iotas {
IOTA_OMIKRON_CONNECTIONS.remove(iota_id);
// Step 2: Remove primary connections safely
for iota_id in primary_to_remove {
IOTA_PRIMARY_OMIKRON_CONNECTION.remove(&iota_id);
}
// Step 3: Remove users that were on this omikron
USER_STATUS_MAP.retain(|_, status| status.omikron_id != omikron_id);
// Step 4: For offline iotas, remove associated users from USER_STATUS_MAP
for iota_id in offline_iotas {
if let Ok(users) = sql::sql::get_users_by_iota_id(iota_id).await {
for user in users {
USER_STATUS_MAP.remove(&user.0);
}
}
// Finally remove the empty connections vector
IOTA_OMIKRON_CONNECTIONS.remove(&iota_id);
}
}

154
src/util/logger.rs
View file

@ -1,4 +1,5 @@
use std::{
collections::HashMap,
fs::{self, OpenOptions},
io::Write,
path::Path,
@ -8,6 +9,8 @@ use std::{
};
use ansi_term::Color;
use epsilon_core::{CommunicationValue, DataTypes, DataValue};
use json::JsonValue;
static LOGGER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
@ -31,8 +34,6 @@ struct LogMessage {
message: String,
}
/// Initialize the logging subsystem.
/// Must be called exactly once during startup.
pub fn startup() {
let (tx, rx) = mpsc::channel::<LogMessage>();
LOGGER.set(tx).expect("Logger already initialized");
@ -62,10 +63,8 @@ pub fn startup() {
let line = format!("{} {} {} {}", ts, sender, msg.prefix, msg.message);
// Console (ANSI-colored)
println!("{}", colorize(msg.kind, msg.is_error).paint(&line));
// File (plain text)
let _ = writeln!(file, "{}", line);
}
});
@ -96,9 +95,6 @@ fn fixed_box(content: &str, width: usize) -> String {
}
}
/** Internal async logging entry point.
* Not exposed publicly; all access goes through macros.
*/
pub fn log_internal(
sender: Option<i64>,
kind: PrintType,
@ -120,7 +116,7 @@ pub fn log_internal(
});
}
}
/// Log a general informational message.
#[macro_export]
macro_rules! log {
// plain
@ -144,7 +140,7 @@ macro_rules! log {
$crate::util::logger::log_internal(None, $kind, "", false, format!($($arg)*))
};
}
/// Log an inbound message (`>`).
#[macro_export]
macro_rules! log_in {
// sender + actor
@ -168,7 +164,7 @@ macro_rules! log_in {
)
};
}
/// Log an outbound message (`<`).
#[macro_export]
macro_rules! log_out {
@ -193,7 +189,7 @@ macro_rules! log_out {
)
};
}
/// Log an error message (`>>`).
#[macro_export]
macro_rules! log_err {
@ -218,3 +214,139 @@ macro_rules! log_err {
)
};
}
// ******** COMMUNICATION VALUES ********
pub fn log_cv_internal(
prefix: &'static str,
cv: &CommunicationValue,
print_type: Option<PrintType>,
) {
let formatted = format_cv(cv);
log_internal(
Some(cv.get_sender() as i64),
print_type.unwrap_or(PrintType::General),
prefix,
false,
formatted,
);
}
pub fn format_cv(cv: &CommunicationValue) -> String {
let mut parts = Vec::new();
let sender = cv.get_sender();
let receiver = cv.get_receiver();
if sender > 0 && receiver > 0 {
parts.push(format!("{} > {}", sender, receiver));
} else if sender > 0 {
parts.push(format!("{}", sender));
} else if receiver > 0 {
parts.push(format!("> {}", receiver));
}
let comm_type = cv.get_type().to_string();
parts.push(format!("{}", comm_type));
let data: &HashMap<DataTypes, DataValue> = cv.get_data_container();
let formated_data =
format_data_container(data.iter().map(|(k, v)| (k.clone(), v.clone())).collect());
parts.push(format!("{}", formated_data));
parts.join(": ")
}
fn format_data_container(data: Vec<(DataTypes, DataValue)>) -> String {
let parts: Vec<String> = data
.into_iter()
.map(|(key, value)| {
let key_str = key.to_string();
match value {
DataValue::Str(s) => format!("{}=\"{}\"", key_str, s),
DataValue::Container(inner) => {
let inner_formatted = format_data_container(inner);
format!("{}={{ {} }}", key_str, inner_formatted)
}
DataValue::Array(arr) => {
let arr_formatted = format_array(arr);
format!("{}=[{}]", key_str, arr_formatted)
}
DataValue::Bool(b) => format!("{}={}", key_str, b),
DataValue::BoolTrue => format!("{}=true", key_str),
DataValue::BoolFalse => format!("{}=false", key_str),
DataValue::Number(num) => format!("{}={}", key_str, num),
_ => "".to_string(),
}
})
.collect();
parts.join(", ")
}
fn format_array(arr: Vec<DataValue>) -> String {
let parts: Vec<String> = arr
.into_iter()
.map(|value| match value {
DataValue::Str(s) => format!("\"{}\"", s),
DataValue::Container(inner) => {
let inner_formatted = format_data_container(inner);
format!("{{ {} }}", inner_formatted)
}
DataValue::Array(inner_arr) => {
let formatted = format_array(inner_arr);
format!("[{}]", formatted)
}
DataValue::Bool(b) => b.to_string(),
DataValue::BoolTrue => "true".to_string(),
DataValue::BoolFalse => "false".to_string(),
DataValue::Number(num) => num.to_string(),
_ => String::new(),
})
.collect();
parts.join(", ")
}
#[macro_export]
macro_rules! log_cv {
($kind:expr, $cv:expr) => {
$crate::util::logger::log_cv_internal("", &$cv, Some($kind))
};
($cv:expr) => {
$crate::util::logger::log_cv_internal("", &$cv, None)
};
}
#[macro_export]
macro_rules! log_cv_in {
($kind:expr, $cv:expr) => {
$crate::util::logger::log_cv_internal("> ", &$cv, Some($kind))
};
($cv:expr) => {
$crate::util::logger::log_cv_internal("> ", &$cv, None)
};
}
#[macro_export]
macro_rules! log_cv_out {
($kind:expr, $cv:expr) => {
$crate::util::logger::log_cv_internal("< ", &$cv, Some($kind))
};
($cv:expr) => {
$crate::util::logger::log_cv_internal("< ", &$cv, None)
};
}