From 10ea9b4ae4dbdf8ebe5110e7370f944b7b11ae71 Mon Sep 17 00:00:00 2001 From: Alex-Emmet Date: Sun, 8 Feb 2026 20:08:26 +0100 Subject: [PATCH] [Add] Notification Logic --- src/data/communication.rs | 6 +++ src/server/omikron_connection.rs | 55 ++++++++++++++++++++ src/sql/sql.rs | 87 +++++++++++++++++++++++++++----- 3 files changed, 135 insertions(+), 13 deletions(-) diff --git a/src/data/communication.rs b/src/data/communication.rs index 7c1660d..1ac5857 100644 --- a/src/data/communication.rs +++ b/src/data/communication.rs @@ -35,6 +35,7 @@ pub enum DataTypes { denied_profiles, content, messages, + notifications, send_time, get_time, get_variant, @@ -139,6 +140,11 @@ pub enum CommunicationType { message_other_iota, message_chunk, messages_get, + + push_notification, + read_notification, + get_notifications, + change_confirm, confirm_receive, confirm_read, diff --git a/src/server/omikron_connection.rs b/src/server/omikron_connection.rs index e8edc41..7ce8f8b 100644 --- a/src/server/omikron_connection.rs +++ b/src/server/omikron_connection.rs @@ -887,6 +887,61 @@ impl OmikronConnection { } return; } + + // NOTIFICATIONS + if cv.is_type(CommunicationType::get_notifications) { + if let Some(JsonValue::Number(user_id)) = cv.get_data(DataTypes::user_id) { + if let Ok(notifications) = + sql::get_notifications(user_id.as_fixed_point_i64(0).unwrap()).await + { + let mut json_array = Vec::new(); + for (sender, amount) in notifications { + let mut obj = JsonValue::new_object(); + let _ = obj.insert("sender", JsonValue::from(sender)); + let _ = obj.insert("amount", JsonValue::from(amount)); + json_array.push(obj); + } + let response = CommunicationValue::new(CommunicationType::get_notifications) + .with_id(cv.get_id()) + .add_array(DataTypes::notifications, json_array); + self.send_message(&response).await; + } + } + } + if cv.is_type(CommunicationType::read_notification) { + if let (Some(JsonValue::Number(user_id)), Some(JsonValue::Number(other_id))) = ( + cv.get_data(DataTypes::receiver_id), + cv.get_data(DataTypes::sender_id), + ) { + if let Ok(_) = sql::read_notification( + user_id.as_fixed_point_i64(0).unwrap(), + other_id.as_fixed_point_i64(0).unwrap(), + ) + .await + { + let response = CommunicationValue::new(CommunicationType::read_notification) + .with_id(cv.get_id()); + self.send_message(&response).await; + } + } + } + if cv.is_type(CommunicationType::push_notification) { + if let (Some(JsonValue::Number(user_id)), Some(JsonValue::Number(other_id))) = ( + cv.get_data(DataTypes::receiver_id), + cv.get_data(DataTypes::sender_id), + ) { + if let Ok(_) = sql::add_notification( + user_id.as_fixed_point_i64(0).unwrap(), + other_id.as_fixed_point_i64(0).unwrap(), + ) + .await + { + let response = CommunicationValue::new(CommunicationType::push_notification) + .with_id(cv.get_id()); + self.send_message(&response).await; + } + } + } } async fn send_error_response(&self, message_id: &Uuid, error_type: CommunicationType) { diff --git a/src/sql/sql.rs b/src/sql/sql.rs index 6a6ffc9..bb79893 100644 --- a/src/sql/sql.rs +++ b/src/sql/sql.rs @@ -94,7 +94,17 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> { ) .execute(&pool) .await; - + let _ = sqlx::query( + "CREATE TABLE IF NOT EXISTS + notifications ( + id BIGINT UNSIGNED NOT NULL PRIMARY KEY AUTO INCREMENT, + sender_id BIGINT UNSIGNED NOT NULL, + receiver_id BIGINT UNSIGNED NOT NULL, + amount BIGINT UNSIGNED NOT NULL DEFAULT 0 + )", + ) + .execute(&pool) + .await; *db_lock = Some(pool); Ok(()) } @@ -396,6 +406,17 @@ pub async fn change_status(id: i64, new_status: String) -> Result<(), sqlx::Erro Ok(()) } +pub async fn delete_user(id: i64) -> Result<(), sqlx::Error> { + let db_lock = SQL_DB.read().await; + let pool = db_lock.as_ref().expect("Database pool is not initialized"); + + sqlx::query("DELETE FROM users WHERE id = CAST(? AS UNSIGNED)") + .bind(id) + .execute(pool) + .await?; + + Ok(()) +} pub async fn change_iota_id(id: i64, new_iota_id: i64) -> Result<(), sqlx::Error> { let db_lock = SQL_DB.read().await; let pool = db_lock.as_ref().expect("Database pool is not initialized"); @@ -524,18 +545,6 @@ pub async fn register_complete_iota(id: i64, public_key: String) -> Result<(), s Ok(()) } -pub async fn delete_user(id: i64) -> Result<(), sqlx::Error> { - let db_lock = SQL_DB.read().await; - let pool = db_lock.as_ref().expect("Database pool is not initialized"); - - sqlx::query("DELETE FROM users WHERE id = CAST(? AS UNSIGNED)") - .bind(id) - .execute(pool) - .await?; - - Ok(()) -} - pub async fn get_iota_by_id(id: i64) -> Result<(i64, String), sqlx::Error> { let db_lock = SQL_DB.read().await; let pool = db_lock.as_ref().expect("Database pool is not initialized"); @@ -607,3 +616,55 @@ pub async fn get_omikron_by_id(id: i64) -> Result<(String, String), sqlx::Error> _ => Err(sqlx::Error::RowNotFound), } } + +// ========================================================================================== +// PHI +// ========================================================================================== + +pub async fn add_notification(sender_id: i64, receiver_id: i64) -> Result<(), sqlx::Error> { + let db_lock = SQL_DB.read().await; + let pool = db_lock.as_ref().expect("Database pool is not initialized"); + + sqlx::query( + r#" + INSERT INTO notifications (sender_id, receiver_id, amount) + VALUES (?, ?, 1) + ON DUPLICATE KEY UPDATE amount = amount + 1 + "#, + ) + .bind(sender_id) + .bind(receiver_id) + .execute(pool) + .await?; + + Ok(()) +} +pub async fn read_notification(sender_id: i64, receiver_id: i64) -> Result<(), sqlx::Error> { + let db_lock = SQL_DB.read().await; + let pool = db_lock.as_ref().expect("Database pool is not initialized"); + + sqlx::query( + r#" + DELETE FROM notifications WHERE sender_id = ? AND receiver_id = ? + "#, + ) + .bind(sender_id) + .bind(receiver_id) + .execute(pool) + .await?; + + Ok(()) +} +pub async fn get_notifications(user_id: i64) -> Result, sqlx::Error> { + let db_lock = SQL_DB.read().await; + let pool = db_lock.as_ref().expect("Database pool is not initialized"); + + sqlx::query_as::<_, (i64, i64)>( + r#" + SELECT sender_id, amount FROM notifications WHERE receiver_id = ? + "#, + ) + .bind(user_id) + .fetch_all(pool) + .await +}