fix: notifs not live (#6299)

* fix: notifs not live

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

Co-authored-by: Sychic <47618543+Sychic@users.noreply.github.com>
Signed-off-by: Calum H. <hendersoncal117@gmail.com>

* fix: fmt

---------

Signed-off-by: Calum H. <hendersoncal117@gmail.com>
Co-authored-by: Sychic <47618543+Sychic@users.noreply.github.com>

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

@@ -204,10 +204,11 @@ impl NotificationBuilder {
         users: Vec<DBUserId>,
         transaction: &mut PgTransaction<'_>,
         redis: &RedisPool,
-    ) -> Result<(), DatabaseError> {
-        self.insert_many_records(&users, transaction).await?;
+    ) -> Result<Vec<DBNotificationId>, DatabaseError> {
+        let notification_ids =
+            self.insert_many_records(&users, transaction).await?;
         DBNotification::clear_user_notifications_cache(&users, redis).await?;
-        Ok(())
+        Ok(notification_ids)
     }
 
     pub async fn insert_many_deliveries(

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

@@ -151,12 +151,42 @@ pub async fn create_email_sync(
         .filter(|id| !already_notified.contains(id))
         .collect::<Vec<_>>();
 
-    NotificationBuilder { body: body.clone() }
+    let notification_ids = NotificationBuilder { body: body.clone() }
         .insert_many_without_delivery(notification_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 {
+            user_id: to_user,
+            id: notification_id,
+            ..
+        } = notification;
+        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"
+            );
+        }
+    }
+
     let mut email_txn = pool.begin().await?;
 
     let mut failed = Vec::new();