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

[Add] Notification Logic

Browse source
This commit is contained in:
Alex-Emmet 2026-02-08 20:08:26 +01:00
commit c4bc26935b
3 changed files with 138 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/data/communication.rs
View file

@ -35,6 +35,7 @@ pub enum DataTypes {
denied_profiles,
content,
messages,
notifications,
send_time,
get_time,
get_variant,
@ -53,6 +54,7 @@ pub enum DataTypes {
signature,
signed,
message,
message_state,
last_ping,
ping_iota,
ping_clients,
@ -112,6 +114,7 @@ impl DataTypes {
#[allow(non_camel_case_types, dead_code)]
pub enum CommunicationType {
error,
error_anonymous,
error_internal,
error_invalid_data,
error_invalid_user_id,
@ -134,11 +137,17 @@ pub enum CommunicationType {
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,

55
src/server/omikron_connection.rs
View file

@ -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) {

87
src/sql/sql.rs
View file

@ -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<Vec<(i64, i64)>, 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
}