feat: add notifs onto friends ws temporarily

apps/app-frontend/src/App.vue

@@ -95,7 +95,7 @@ import { hide_ads_window, init_ads_window, show_ads_window } from '@/helpers/ads
 import { debugAnalytics, initAnalytics, trackEvent } from '@/helpers/analytics'
 import { check_reachable } from '@/helpers/auth.js'
 import { get_user, get_version } from '@/helpers/cache.js'
-import { command_listener, warning_listener } from '@/helpers/events.js'
+import { command_listener, notification_listener, warning_listener } from '@/helpers/events.js'
 import { cancelLogin, get as getCreds, login, logout } from '@/helpers/mr_auth.ts'
 import { create_profile_and_install_from_file } from '@/helpers/pack'
 import { list } from '@/helpers/profile.js'
@@ -241,6 +241,7 @@ const {
 
 const news = ref([])
 const availableSurvey = ref(false)
+const displayedServerInviteNotifications = new Set()
 
 const offline = ref(!navigator.onLine)
 window.addEventListener('offline', () => {
@@ -752,6 +753,79 @@ const accounts = ref(null)
 provide('accountsCard', accounts)
 
 command_listener(handleCommand)
+notification_listener(handleLiveNotification)
+
+async function markLiveNotificationRead(notification) {
+	await tauriApiClient.labrinth.notifications_v2.markAsRead(notification.id)
+}
+
+async function respondToServerInvite(notification, action) {
+	const serverId = notification.body?.server_id
+	if (typeof serverId !== 'string') {
+		throw new Error('Missing server ID for invite notification.')
+	}
+
+	await markLiveNotificationRead(notification)
+	await tauriApiClient.request(`/servers/${serverId}/invites/${action}`, {
+		api: 'archon',
+		version: 1,
+		method: 'POST',
+	})
+
+	return serverId
+}
+
+async function acceptServerInviteNotification(notification) {
+	try {
+		const serverId = await respondToServerInvite(notification, 'accept')
+		await router.push(`/hosting/manage/${encodeURIComponent(serverId)}`)
+		queryClient.invalidateQueries({ queryKey: ['servers'] })
+	} catch (error) {
+		handleError(error)
+	}
+}
+
+async function declineServerInviteNotification(notification) {
+	try {
+		await respondToServerInvite(notification, 'decline')
+	} catch (error) {
+		handleError(error)
+	}
+}
+
+async function handleLiveNotification(notification) {
+	if (notification?.body?.type !== 'server_invite' || notification.read) return
+	if (displayedServerInviteNotifications.has(notification.id)) return
+
+	displayedServerInviteNotifications.add(notification.id)
+
+	const serverName =
+		typeof notification.body.server_name === 'string' ? notification.body.server_name : 'a server'
+	const inviterId = notification.body.invited_by
+	const invitedBy =
+		typeof inviterId === 'string' ? await get_user(inviterId, 'bypass').catch(() => null) : null
+	const inviterName = invitedBy?.username ?? 'Someone'
+
+	addPopupNotification({
+		title: 'Modrinth Hosting',
+		text: `${inviterName} has invited you to join ${serverName}.`,
+		type: 'info',
+		buttons: [
+			{
+				label: 'Accept',
+				action: () => acceptServerInviteNotification(notification),
+				color: 'brand',
+			},
+			{
+				label: 'Decline',
+				action: () => declineServerInviteNotification(notification),
+				color: 'red',
+			},
+		],
+		autoCloseMs: null,
+	})
+}
+
 async function handleCommand(e) {
 	if (!e) return
 

apps/app-frontend/src/helpers/events.js

@@ -98,6 +98,10 @@ export async function friend_listener(callback) {
 	return await listen('friend', (event) => callback(event.payload))
 }
 
+export async function notification_listener(callback) {
+	return await listen('notification', (event) => callback(event.payload))
+}
+
 /// Payload for the 'log' event
 /*
     LogPayload {

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,29 @@ pub async fn create(
         ));
     }
 
-    NotificationBuilder { body }
+    let notification_ids = NotificationBuilder { body }
         .insert_many(user_ids, &mut txn, &redis)
         .await?;
 
     txn.commit().await?;
 
+    let notifications = DBNotification::get_many(&notification_ids, &**pool)
+        .await?
+        .into_iter()
+        .map(Notification::from)
+        .collect::<Vec<_>>();
+
+    for notification in notifications {
+        broadcast_friends_message(
+            &redis,
+            RedisFriendsMessage::Notification {
+                to_user: notification.user_id,
+                notification,
+            },
+        )
+        .await?;
+    }
+
     Ok(HttpResponse::Accepted().finish())
 }
 

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

@@ -3,6 +3,7 @@ use crate::auth::validate::get_user_record_from_bearer_token;
 use crate::database::PgPool;
 use crate::database::models::friend_item::DBFriend;
 use crate::database::redis::RedisPool;
+use crate::models::notifications::Notification;
 use crate::models::pats::Scopes;
 use crate::models::users::User;
 use crate::queue::session::AuthQueue;
@@ -42,6 +43,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,
@@ -449,6 +451,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 },
+    StatusUpdate {
-    UserOffline { user: UserId },
+        status: UserStatus,
-    DirectStatusUpdate { to_user: UserId, 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(_) => {}
             }
         });

packages/api-client/src/modules/labrinth/types.ts

@@ -1648,6 +1648,8 @@ export namespace Labrinth {
 				message_id?: string
 				invited_by?: string
 				organization_id?: string
+				server_id?: string
+				server_name?: string
 				team_id?: string
 				role?: string
 				old_status?: string

packages/app-lib/src/event/emit.rs

@@ -8,6 +8,7 @@ use crate::event::{
     LoadingPayload, ProcessPayload, ProfilePayload, WarningPayload,
 };
 use futures::prelude::*;
+use serde_json::Value;
 #[cfg(feature = "tauri")]
 use tauri::{Emitter, Manager};
 use uuid::Uuid;
@@ -303,6 +304,20 @@ pub async fn emit_friend(payload: FriendPayload) -> crate::Result<()> {
     Ok(())
 }
 
+#[allow(unused_variables)]
+pub async fn emit_notification(payload: Value) -> crate::Result<()> {
+    #[cfg(feature = "tauri")]
+    {
+        let event_state = crate::EventState::get()?;
+        event_state
+            .app
+            .emit("notification", payload)
+            .map_err(EventError::from)?;
+    }
+
+    Ok(())
+}
+
 // loading_join! macro
 // loading_join!(key: Option<&LoadingBarId>, total: f64, message: Option<&str>; task1, task2, task3...)
 // This will submit a loading event with the given message for each task as they complete

packages/app-lib/src/state/friends.rs

@@ -1,7 +1,7 @@
 use crate::ErrorKind;
 use crate::data::ModrinthCredentials;
 use crate::event::FriendPayload;
-use crate::event::emit::emit_friend;
+use crate::event::emit::{emit_friend, emit_notification};
 use crate::state::tunnel::InternalTunnelSocket;
 use crate::state::{ProcessManager, Profile, TunnelSocket};
 use crate::util::fetch::{FetchSemaphore, fetch_advanced, fetch_json};
@@ -22,6 +22,7 @@ use futures::{SinkExt, StreamExt};
 use reqwest::Method;
 use reqwest::header::HeaderValue;
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use std::net::SocketAddr;
 use std::ops::Deref;
 use std::sync::Arc;
@@ -120,16 +121,34 @@ impl FriendsSocket {
                                 Ok(msg) => {
                                     let server_message = match msg {
                                         Message::Text(text) => {
-                                            ServerToClientMessage::deserialize(
+                                            match ServerToClientMessage::deserialize(
                                                 Either::Left(&text),
-                                            )
+                                            ) {
-                                            .ok()
+                                                Ok(message) => Some(message),
+                                                Err(_) => {
+                                                    if let Ok(notification) =
+                                                        serde_json::from_str::<Value>(&text)
+                                                    {
+                                                        let _ = Self::handle_notification(notification).await;
+                                                    }
+                                                    None
+                                                }
+                                            }
                                         }
                                         Message::Binary(bytes) => {
-                                            ServerToClientMessage::deserialize(
+                                            match ServerToClientMessage::deserialize(
                                                 Either::Right(&bytes),
-                                            )
+                                            ) {
-                                            .ok()
+                                                Ok(message) => Some(message),
+                                                Err(_) => {
+                                                    if let Ok(notification) =
+                                                        serde_json::from_slice::<Value>(&bytes)
+                                                    {
+                                                        let _ = Self::handle_notification(notification).await;
+                                                    }
+                                                    None
+                                                }
+                                            }
                                         }
                                         Message::Ping(bytes) => {
                                             if let Some(write) = write_handle
@@ -224,6 +243,19 @@ impl FriendsSocket {
         Ok(())
     }
 
+    async fn handle_notification(notification: Value) -> crate::Result<()> {
+        if notification
+            .get("body")
+            .and_then(|body| body.get("type"))
+            .and_then(Value::as_str)
+            .is_some()
+        {
+            emit_notification(notification).await?;
+        }
+
+        Ok(())
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn socket_loop() -> crate::Result<()> {
         let state = crate::State::get().await?;