feat: add notifs onto friends ws temporarily (#6290)

* feat: add notifs onto friends ws temporarily

* fix: lint + styling

* fix: regressions

apps/labrinth/src/database/models/notification_item.rs

@@ -40,7 +40,8 @@ impl NotificationBuilder {
         transaction: &mut PgTransaction<'_>,
         redis: &RedisPool,
     ) -> Result<(), DatabaseError> {
-        self.insert_many(vec![user], transaction, redis).await
+        self.insert_many(vec![user], transaction, redis).await?;
+        Ok(())
     }
 
     pub async fn insert_many_payout_notifications(
@@ -133,7 +134,7 @@ impl NotificationBuilder {
         &self,
         users: &[DBUserId],
         transaction: &mut PgTransaction<'_>,
-    ) -> Result<Vec<i64>, DatabaseError> {
+    ) -> Result<Vec<DBNotificationId>, DatabaseError> {
         let notification_ids =
             generate_many_notification_ids(users.len(), &mut *transaction)
                 .await?;
@@ -145,7 +146,7 @@ impl NotificationBuilder {
             .collect::<Vec<_>>();
 
         let users_raw_ids = users.iter().map(|x| x.0).collect::<Vec<_>>();
-        let notification_ids =
+        let notification_ids_raw =
             notification_ids.iter().map(|x| x.0).collect::<Vec<_>>();
 
         sqlx::query!(
@@ -155,7 +156,7 @@ impl NotificationBuilder {
             )
             SELECT * FROM UNNEST($1::bigint[], $2::bigint[], $3::jsonb[])
             ",
-            &notification_ids[..],
+            &notification_ids_raw[..],
             &users_raw_ids[..],
             &bodies[..],
         )
@@ -170,11 +171,13 @@ impl NotificationBuilder {
         users: Vec<DBUserId>,
         transaction: &mut PgTransaction<'_>,
         redis: &RedisPool,
-    ) -> Result<(), DatabaseError> {
+    ) -> Result<Vec<DBNotificationId>, DatabaseError> {
         let notification_ids =
             self.insert_many_records(&users, transaction).await?;
 
         let users_raw_ids = users.iter().map(|x| x.0).collect::<Vec<_>>();
+        let notification_ids_raw =
+            notification_ids.iter().map(|x| x.0).collect::<Vec<_>>();
 
         let notification_types = notification_ids
             .iter()
@@ -184,14 +187,14 @@ impl NotificationBuilder {
         NotificationBuilder::insert_many_deliveries(
             transaction,
             redis,
-            &notification_ids,
+            &notification_ids_raw,
             &users_raw_ids,
             &notification_types,
             &users,
         )
         .await?;
 
-        Ok(())
+        Ok(notification_ids)
     }
 
     /// Like [`insert_many`], but skips queuing deliveries so the caller can

apps/labrinth/src/routes/internal/external_notifications.rs

@@ -7,12 +7,14 @@ use crate::database::models::user_item::DBUser;
 use crate::database::redis::RedisPool;
 use crate::models::users::Role;
 use crate::models::v3::notifications::{
-    NotificationBody, NotificationDeliveryStatus,
+    Notification, NotificationBody, NotificationDeliveryStatus,
 };
 use crate::models::v3::pats::Scopes;
 use crate::queue::email::EmailQueue;
 use crate::queue::session::AuthQueue;
 use crate::routes::ApiError;
+use crate::routes::internal::statuses::broadcast_friends_message;
+use crate::sync::friends::RedisFriendsMessage;
 use crate::util::guards::external_notification_key_guard;
 use actix_web::http::StatusCode;
 use actix_web::web;
@@ -58,12 +60,39 @@ pub async fn create(
         ));
     }
 
-    NotificationBuilder { body }
+    let notification_ids = NotificationBuilder { body }
         .insert_many(user_ids, &mut txn, &redis)
         .await?;
 
+    let notifications = DBNotification::get_many(&notification_ids, &mut txn)
+        .await?
+        .into_iter()
+        .map(Notification::from)
+        .collect::<Vec<_>>();
+
     txn.commit().await?;
 
+    for notification in notifications {
+        let notification_id = notification.id;
+        let to_user = notification.user_id;
+        if let Err(error) = broadcast_friends_message(
+            &redis,
+            RedisFriendsMessage::Notification {
+                to_user,
+                notification,
+            },
+        )
+        .await
+        {
+            tracing::warn!(
+                ?error,
+                ?notification_id,
+                ?to_user,
+                "failed to broadcast realtime notification"
+            );
+        }
+    }
+
     Ok(HttpResponse::Accepted().finish())
 }
 

apps/labrinth/src/routes/internal/statuses.rs

@@ -2,7 +2,9 @@ use crate::auth::AuthenticationError;
 use crate::auth::validate::get_user_record_from_bearer_token;
 use crate::database::PgPool;
 use crate::database::models::friend_item::DBFriend;
+use crate::database::models::notification_item::DBNotification;
 use crate::database::redis::RedisPool;
+use crate::models::notifications::{Notification, NotificationBody};
 use crate::models::pats::Scopes;
 use crate::models::users::User;
 use crate::queue::session::AuthQueue;
@@ -42,6 +44,7 @@ struct LauncherHeartbeatInit {
     code: String,
 }
 
+// TODO: Move launcher-specific tunnel traffic to a proper launcher websocket endpoint.
 #[get("launcher_socket")]
 pub async fn ws_init(
     req: HttpRequest,
@@ -127,6 +130,26 @@ pub async fn ws_init(
         )?)
         .await;
 
+    let unread_server_invites = DBNotification::get_many_user_exposed_on_site(
+        user_id.into(),
+        &**pool,
+        &redis,
+    )
+    .await?
+    .into_iter()
+    .filter(|notification| {
+        !notification.read
+            && matches!(
+                &notification.body,
+                NotificationBody::ServerInvite { .. }
+            )
+    })
+    .map(Notification::from);
+
+    for notification in unread_server_invites {
+        let _ = session.text(serde_json::to_string(&notification)?).await;
+    }
+
     let db = db.clone();
     let socket_id = db.next_socket_id.fetch_add(1, Ordering::Relaxed);
     db.sockets
@@ -449,6 +472,25 @@ pub async fn send_message_to_user(
     Ok(())
 }
 
+pub async fn send_notification_to_user(
+    db: &ActiveSockets,
+    user: UserId,
+    notification: &Notification,
+) -> Result<(), crate::database::models::DatabaseError> {
+    let message = serde_json::to_string(notification)?;
+
+    if let Some(socket_ids) = db.sockets_by_user_id.get(&user) {
+        for socket_id in socket_ids.iter() {
+            if let Some(socket) = db.sockets.get(&socket_id) {
+                let mut socket = socket.socket.clone();
+                let _ = socket.text(message.clone()).await;
+            }
+        }
+    }
+
+    Ok(())
+}
+
 pub async fn close_socket(
     id: SocketId,
     pool: &PgPool,

apps/labrinth/src/sync/friends.rs

@@ -1,7 +1,8 @@
 use crate::database::PgPool;
+use crate::models::notifications::Notification;
 use crate::queue::socket::ActiveSockets;
 use crate::routes::internal::statuses::{
-    broadcast_to_local_friends, send_message_to_user,
+    broadcast_to_local_friends, send_message_to_user, send_notification_to_user,
 };
 use actix_web::web::Data;
 use ariadne::ids::UserId;
@@ -14,12 +15,23 @@ use tokio_stream::StreamExt;
 
 pub const FRIENDS_CHANNEL_NAME: &str = "friends";
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
 pub enum RedisFriendsMessage {
-    StatusUpdate { status: UserStatus },
-    UserOffline { user: UserId },
-    DirectStatusUpdate { to_user: UserId, status: UserStatus },
+    StatusUpdate {
+        status: UserStatus,
+    },
+    UserOffline {
+        user: UserId,
+    },
+    DirectStatusUpdate {
+        to_user: UserId,
+        status: UserStatus,
+    },
+    Notification {
+        to_user: UserId,
+        notification: Notification,
+    },
 }
 
 impl ToRedisArgs for RedisFriendsMessage {
@@ -80,6 +92,18 @@ pub async fn handle_pubsub(
                     .await;
                 }
 
+                Ok(RedisFriendsMessage::Notification {
+                    to_user,
+                    notification,
+                }) => {
+                    let _ = send_notification_to_user(
+                        &sockets,
+                        to_user,
+                        &notification,
+                    )
+                    .await;
+                }
+
                 Err(_) => {}
             }
         });