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

consistent Data in SQL, better Connections

Browse source
This commit is contained in:
Alex Emmet 2026-05-23 02:44:24 +02:00
commit 418deca3b6
3 changed files with 50 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

60
src/sql/sql.rs
View file

@ -53,7 +53,7 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
let _ = sqlx::query( let _ = sqlx::query(
"CREATE TABLE IF NOT EXISTS "CREATE TABLE IF NOT EXISTS
users ( users (
id BIGINT UNSIGNED NOT NULL PRIMARY KEY, id BIGINT NOT NULL PRIMARY KEY,
username VARCHAR(15) NOT NULL UNIQUE COLLATE utf8mb4_bin, username VARCHAR(15) NOT NULL UNIQUE COLLATE utf8mb4_bin,
display VARCHAR(15) COLLATE utf8mb4_bin, display VARCHAR(15) COLLATE utf8mb4_bin,
status VARCHAR(15) COLLATE utf8mb4_bin, status VARCHAR(15) COLLATE utf8mb4_bin,
@ -63,8 +63,8 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
sub_end BIGINT(20) NOT NULL DEFAULT 0, sub_end BIGINT(20) NOT NULL DEFAULT 0,
public_key TEXT NOT NULL COLLATE utf8mb4_bin, public_key TEXT NOT NULL COLLATE utf8mb4_bin,
private_key_hash TEXT NOT NULL COLLATE utf8mb4_bin DEFAULT '', private_key_hash TEXT NOT NULL COLLATE utf8mb4_bin DEFAULT '',
iota_id BIGINT UNSIGNED NOT NULL, iota_id BIGINT NOT NULL,
token VARCHAR(255) NOT NULL UNIQUE COLLATE utf8mb4_bin token VARCHAR(256) NOT NULL UNIQUE COLLATE utf8mb4_bin
)", )",
) )
.execute(&pool) .execute(&pool)
@ -72,7 +72,7 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
let _ = sqlx::query( let _ = sqlx::query(
"CREATE TABLE IF NOT EXISTS "CREATE TABLE IF NOT EXISTS
iotas ( iotas (
id BIGINT UNSIGNED NOT NULL PRIMARY KEY, id BIGINT NOT NULL PRIMARY KEY,
public_key VARCHAR(255) NOT NULL COLLATE utf8mb4_bin public_key VARCHAR(255) NOT NULL COLLATE utf8mb4_bin
)", )",
) )
@ -81,7 +81,7 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
let _ = sqlx::query( let _ = sqlx::query(
"CREATE TABLE IF NOT EXISTS "CREATE TABLE IF NOT EXISTS
omikrons ( omikrons (
id BIGINT UNSIGNED NOT NULL PRIMARY KEY, id BIGINT NOT NULL PRIMARY KEY,
public_key VARCHAR(255) NOT NULL COLLATE utf8mb4_bin, public_key VARCHAR(255) NOT NULL COLLATE utf8mb4_bin,
location VARCHAR(255) NOT NULL COLLATE utf8mb4_bin, location VARCHAR(255) NOT NULL COLLATE utf8mb4_bin,
ip_address VARCHAR(255) NOT NULL COLLATE utf8mb4_bin ip_address VARCHAR(255) NOT NULL COLLATE utf8mb4_bin
@ -92,10 +92,10 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
let _ = sqlx::query( let _ = sqlx::query(
"CREATE TABLE IF NOT EXISTS "CREATE TABLE IF NOT EXISTS
notifications ( notifications (
id BIGINT UNSIGNED NOT NULL PRIMARY KEY AUTO INCREMENT, id BIGINT NOT NULL PRIMARY KEY AUTO_INCREMENT,
sender_id BIGINT UNSIGNED NOT NULL, sender_id BIGINT NOT NULL,
receiver_id BIGINT UNSIGNED NOT NULL, receiver_id BIGINT NOT NULL,
amount BIGINT UNSIGNED NOT NULL DEFAULT 0 amount BIGINT NOT NULL DEFAULT 0
)", )",
) )
.execute(&pool) .execute(&pool)
@ -243,7 +243,7 @@ pub async fn get_by_user_id(
}; };
let row = sqlx::query( let row = sqlx::query(
"SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, private_key_hash, token FROM users WHERE id = CAST(? AS UNSIGNED)", "SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, private_key_hash, token FROM users WHERE id = ?",
) )
.bind(id) .bind(id)
.fetch_optional(&pool) .fetch_optional(&pool)
@ -311,7 +311,7 @@ pub async fn get_users_by_iota_id(
}; };
let rows = sqlx::query( let rows = sqlx::query(
"SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, private_key_hash, token FROM users WHERE iota_id = CAST(? AS UNSIGNED)", "SELECT id, iota_id, username, display, status, about, avatar, sub_level, sub_end, public_key, private_key_hash, token FROM users WHERE iota_id = ?",
) )
.bind(iota_id_param) .bind(iota_id_param)
.fetch_all(&pool) .fetch_all(&pool)
@ -360,7 +360,7 @@ pub async fn change_username(id: i64, new_username: String) -> Result<(), sqlx::
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET username = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET username = ? WHERE id = ?")
.bind(new_username) .bind(new_username)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -378,7 +378,7 @@ pub async fn change_display_name(id: i64, new_display: String) -> Result<(), sql
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET display = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET display = ? WHERE id = ?")
.bind(new_display) .bind(new_display)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -396,7 +396,7 @@ pub async fn change_avatar(id: i64, new_avatar: String) -> Result<(), sqlx::Erro
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET avatar = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET avatar = ? WHERE id = ?")
.bind(new_avatar) .bind(new_avatar)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -414,7 +414,7 @@ pub async fn change_about(id: i64, new_about: String) -> Result<(), sqlx::Error>
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET about = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET about = ? WHERE id = ?")
.bind(new_about) .bind(new_about)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -432,7 +432,7 @@ pub async fn change_status(id: i64, new_status: String) -> Result<(), sqlx::Erro
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET status = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET status = ? WHERE id = ?")
.bind(new_status) .bind(new_status)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -450,7 +450,7 @@ pub async fn delete_user(id: i64) -> Result<(), sqlx::Error> {
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("DELETE FROM users WHERE id = CAST(? AS UNSIGNED)") sqlx::query("DELETE FROM users WHERE id = ?")
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
.await?; .await?;
@ -466,7 +466,7 @@ pub async fn change_iota_id(id: i64, new_iota_id: i64) -> Result<(), sqlx::Error
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET iota_id = CAST(? AS UNSIGNED) WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET iota_id = ? WHERE id = ?")
.bind(new_iota_id) .bind(new_iota_id)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -488,9 +488,7 @@ pub async fn change_keys(
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query( sqlx::query("UPDATE users SET public_key = ?, private_key_hash = ? WHERE id = ?")
"UPDATE users SET public_key = ?, private_key_hash = ? WHERE id = CAST(? AS UNSIGNED)",
)
.bind(new_public_key) .bind(new_public_key)
.bind(new_private_key_hash) .bind(new_private_key_hash)
.bind(id) .bind(id)
@ -508,7 +506,7 @@ pub async fn change_token(id: i64, new_token: String) -> Result<(), sqlx::Error>
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE users SET token = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE users SET token = ? WHERE id = ?")
.bind(new_token) .bind(new_token)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -624,19 +622,17 @@ pub async fn get_iota_by_id(id: i64) -> Result<(i64, String), sqlx::Error> {
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
let result = sqlx::query_as::<_, (u64, Vec<u8>)>( let result =
"SELECT id, public_key FROM iotas WHERE id = CAST(? AS UNSIGNED)", sqlx::query_as::<_, (i64, Vec<u8>)>("SELECT id, public_key FROM iotas WHERE id = ?")
)
.bind(id) .bind(id)
.fetch_optional(&pool) .fetch_optional(&pool)
.await; .await;
match result { match result {
Ok(optional_row) => match optional_row { Ok(optional_row) => match optional_row {
Some((id_u64, public_key)) => Ok(( Some((id_i64, public_key)) => {
id_u64 as i64, Ok((id_i64, String::from_utf8_lossy(&public_key).to_string()))
String::from_utf8_lossy(&public_key).to_string(), }
)),
_ => Err(sqlx::Error::RowNotFound), _ => Err(sqlx::Error::RowNotFound),
}, },
Err(e) => Err(e), Err(e) => Err(e),
@ -652,7 +648,7 @@ pub async fn change_iota_key(id: i64, new_key: String) -> Result<(), sqlx::Error
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("UPDATE iotas SET public_key = ? WHERE id = CAST(? AS UNSIGNED)") sqlx::query("UPDATE iotas SET public_key = ? WHERE id = ?")
.bind(new_key) .bind(new_key)
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
@ -670,7 +666,7 @@ pub async fn delete_iota(id: i64) -> Result<(), sqlx::Error> {
.expect("Database pool not initialized") .expect("Database pool not initialized")
}; };
sqlx::query("DELETE FROM iotas WHERE id = CAST(? AS UNSIGNED)") sqlx::query("DELETE FROM iotas WHERE id = ?")
.bind(id) .bind(id)
.execute(&pool) .execute(&pool)
.await?; .await?;
@ -692,7 +688,7 @@ pub async fn get_omikron_by_id(id: i64) -> Result<(String, String), sqlx::Error>
}; };
let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>)>( let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>)>(
"SELECT public_key, ip_address FROM omikrons WHERE id = CAST(? AS UNSIGNED)", "SELECT public_key, ip_address FROM omikrons WHERE id = ?",
) )
.bind(id) .bind(id)
.fetch_optional(&pool) .fetch_optional(&pool)

4
src/sql/user_online_tracker.rs
View file

@ -133,9 +133,9 @@ pub async fn untrack_omikron(omikron_id: i64) {
// Step 4: For offline iotas, remove associated users from USER_STATUS_MAP // Step 4: For offline iotas, remove associated users from USER_STATUS_MAP
for iota_id in offline_iotas { for iota_id in offline_iotas {
if let Ok(users) = sql::sql::get_users_by_iota_id(iota_id).await { if let Ok(users) = sql::sql::get_users_by_iota_id(iota_id.try_into().unwrap()).await {
for user in users { for user in users {
USER_STATUS_MAP.remove(&user.0); USER_STATUS_MAP.remove(&(user.0 as i64));
} }
} }
// Finally remove the empty connections vector // Finally remove the empty connections vector

19
src/transport/omikron_connection.rs
View file

@ -419,8 +419,8 @@ impl OmikronConnection {
log_in!(PrintType::Omega, "User connected"); log_in!(PrintType::Omega, "User connected");
if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() { if let Some(user_id) = cv.get_data(DataTypes::user_id).as_number() {
user_online_tracker::track_user_status( user_online_tracker::track_user_status(
user_id as i64, user_id.try_into().unwrap(),
UserStatus::user_online, UserStatus::user_offline,
omikron_id, omikron_id,
); );
} }
@ -448,11 +448,11 @@ impl OmikronConnection {
user_online_tracker::track_iota_connection(iota_id, omikron_id, true); user_online_tracker::track_iota_connection(iota_id, omikron_id, true);
let mut user_ids = Vec::new(); let mut user_ids = Vec::new();
if let Ok(users) = sql::get_users_by_iota_id(iota_id).await { if let Ok(users) = sql::get_users_by_iota_id(iota_id.try_into().unwrap()).await {
for (user_id, _, _, _, _, _, _, _, _, _, _, _) in users { for (user_id, _, _, _, _, _, _, _, _, _, _, _) in users {
user_ids.push(DataValue::Number(user_id)); user_ids.push(DataValue::Number(user_id.try_into().unwrap()));
user_online_tracker::track_user_status( user_online_tracker::track_user_status(
user_id, user_id.try_into().unwrap(),
UserStatus::user_offline, UserStatus::user_offline,
omikron_id, omikron_id,
); );
@ -476,8 +476,8 @@ impl OmikronConnection {
}; };
let iota_offline = user_online_tracker::untrack_iota_connection(iota_id, omikron_id); let iota_offline = user_online_tracker::untrack_iota_connection(iota_id, omikron_id);
if iota_offline { if iota_offline {
if let Ok(users) = sql::get_users_by_iota_id(iota_id).await { if let Ok(users) = sql::get_users_by_iota_id(iota_id.try_into().unwrap()).await {
let user_ids: Vec<i64> = users.iter().map(|u| u.0).collect(); let user_ids: Vec<i64> = users.iter().map(|u| u.0.try_into().unwrap()).collect();
user_online_tracker::untrack_many_users(&user_ids); user_online_tracker::untrack_many_users(&user_ids);
} }
} }
@ -489,7 +489,7 @@ impl OmikronConnection {
if let DataValue::Number(user_id) = user_id_val { if let DataValue::Number(user_id) = user_id_val {
user_online_tracker::track_user_status( user_online_tracker::track_user_status(
*user_id, *user_id,
UserStatus::user_online, UserStatus::user_offline,
omikron_id, omikron_id,
); );
} }
@ -911,7 +911,8 @@ impl OmikronConnection {
.with_id(cv.get_id()); .with_id(cv.get_id());
self.send(&response).await self.send(&response).await
} else { } else {
let response = CommunicationValue::new(CommunicationType::error_internal) let response =
CommunicationValue::new(CommunicationType::error_internal)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data(DataTypes::error_type, DataValue::Str(error_message)); .add_data(DataTypes::error_type, DataValue::Str(error_message));
self.send(&response).await self.send(&response).await