fix: websocket friends performance

This commit is contained in:
Michael H.
2026-08-30 12:14:27 +02:00
parent aebbe3b2f5
commit 17e3e95645
4 changed files with 21 additions and 17 deletions
@@ -0,0 +1 @@
CREATE INDEX friends_friend_id ON friends (friend_id);
+2 -2
View File
@@ -265,11 +265,11 @@ pub fn app_setup(
let active_sockets = web::Data::new(ActiveSockets::default()); let active_sockets = web::Data::new(ActiveSockets::default());
{ {
let pool = pool.clone(); let ro_pool = ro_pool.clone();
let pubsub_messages = redis_pool.subscribe(FRIENDS_CHANNEL_NAME); let pubsub_messages = redis_pool.subscribe(FRIENDS_CHANNEL_NAME);
let sockets = active_sockets.clone(); let sockets = active_sockets.clone();
actix_rt::spawn(async move { actix_rt::spawn(async move {
handle_pubsub(pubsub_messages, pool, sockets).await; handle_pubsub(pubsub_messages, ro_pool, sockets).await;
}); });
} }
+11 -9
View File
@@ -1,8 +1,8 @@
use crate::auth::AuthenticationError; use crate::auth::AuthenticationError;
use crate::auth::validate::get_user_record_from_bearer_token; 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::friend_item::DBFriend;
use crate::database::models::notification_item::DBNotification; use crate::database::models::notification_item::DBNotification;
use crate::database::{PgPool, ReadOnlyPgPool};
use crate::models::notifications::{Notification, NotificationBody}; use crate::models::notifications::{Notification, NotificationBody};
use crate::models::pats::Scopes; use crate::models::pats::Scopes;
use crate::models::users::User; use crate::models::users::User;
@@ -54,6 +54,7 @@ struct LauncherHeartbeatInit {
pub async fn ws_init( pub async fn ws_init(
req: HttpRequest, req: HttpRequest,
pool: Data<PgPool>, pool: Data<PgPool>,
ro_pool: Data<ReadOnlyPgPool>,
web::Query(auth): web::Query<LauncherHeartbeatInit>, web::Query(auth): web::Query<LauncherHeartbeatInit>,
body: Payload, body: Payload,
db: Data<ActiveSockets>, db: Data<ActiveSockets>,
@@ -93,7 +94,7 @@ pub async fn ws_init(
}; };
let friends = let friends =
DBFriend::get_user_friends(user.id.into(), Some(true), &**pool) DBFriend::get_user_friends(user.id.into(), Some(true), &***ro_pool)
.await .await
.wrap_internal_err("fetching friends from database")?; .wrap_internal_err("fetching friends from database")?;
@@ -140,7 +141,7 @@ pub async fn ws_init(
let unread_launcher_invites = let unread_launcher_invites =
DBNotification::get_many_user_exposed_on_site( DBNotification::get_many_user_exposed_on_site(
user_id.into(), user_id.into(),
&**pool, &***ro_pool,
&redis, &redis,
) )
.await .await
@@ -336,7 +337,7 @@ pub async fn ws_init(
let _ = broadcast_to_local_friends( let _ = broadcast_to_local_friends(
user.id, user.id,
ServerToClientMessage::FriendSocketStoppedListening { user: user.id }, ServerToClientMessage::FriendSocketStoppedListening { user: user.id },
&pool, &ro_pool,
&db, &db,
) )
.await; .await;
@@ -394,7 +395,7 @@ pub async fn ws_init(
} }
let _ = shutdown_sender.send(()); let _ = shutdown_sender.send(());
let _ = close_socket(socket_id, &pool, &db, &redis).await; let _ = close_socket(socket_id, &ro_pool, &db, &redis).await;
}); });
Ok(res) Ok(res)
@@ -413,14 +414,15 @@ pub async fn broadcast_friends_message(
pub async fn broadcast_to_local_friends( pub async fn broadcast_to_local_friends(
user_id: UserId, user_id: UserId,
message: ServerToClientMessage, message: ServerToClientMessage,
pool: &PgPool, ro_pool: &ReadOnlyPgPool,
sockets: &ActiveSockets, sockets: &ActiveSockets,
) -> Result<(), crate::database::models::DatabaseError> { ) -> Result<(), crate::database::models::DatabaseError> {
broadcast_to_known_local_friends( broadcast_to_known_local_friends(
user_id, user_id,
message, message,
sockets, sockets,
DBFriend::get_user_friends(user_id.into(), Some(true), pool).await?, DBFriend::get_user_friends(user_id.into(), Some(true), &**ro_pool)
.await?,
) )
.await .await
} }
@@ -508,7 +510,7 @@ pub async fn send_notification_to_user(
pub async fn close_socket( pub async fn close_socket(
id: SocketId, id: SocketId,
pool: &PgPool, ro_pool: &ReadOnlyPgPool,
db: &ActiveSockets, db: &ActiveSockets,
redis: &RedisPool, redis: &RedisPool,
) -> Result<(), crate::database::models::DatabaseError> { ) -> Result<(), crate::database::models::DatabaseError> {
@@ -541,7 +543,7 @@ pub async fn close_socket(
ServerToClientMessage::SocketClosed { ServerToClientMessage::SocketClosed {
socket: owned_socket, socket: owned_socket,
}, },
pool, ro_pool,
db, db,
) )
.await; .await;
+7 -6
View File
@@ -1,4 +1,4 @@
use crate::database::PgPool; use crate::database::ReadOnlyPgPool;
use crate::database::models::notification_item::DBNotification; use crate::database::models::notification_item::DBNotification;
use crate::models::ids::NotificationId; use crate::models::ids::NotificationId;
use crate::models::notifications::Notification; use crate::models::notifications::Notification;
@@ -47,13 +47,13 @@ impl ToSingleRedisArg for RedisFriendsMessage {}
pub async fn handle_pubsub( pub async fn handle_pubsub(
mut messages: mpsc::Receiver<Vec<u8>>, mut messages: mpsc::Receiver<Vec<u8>>,
pool: PgPool, ro_pool: ReadOnlyPgPool,
sockets: Data<ActiveSockets>, sockets: Data<ActiveSockets>,
) { ) {
while let Some(message) = messages.recv().await { while let Some(message) = messages.recv().await {
let payload = postcard::from_bytes::<RedisFriendsMessage>(&message); let payload = postcard::from_bytes::<RedisFriendsMessage>(&message);
let pool = pool.clone(); let ro_pool = ro_pool.clone();
let sockets = sockets.clone(); let sockets = sockets.clone();
actix_rt::spawn(async move { actix_rt::spawn(async move {
match payload { match payload {
@@ -61,7 +61,7 @@ pub async fn handle_pubsub(
let _ = broadcast_to_local_friends( let _ = broadcast_to_local_friends(
status.user_id, status.user_id,
ServerToClientMessage::StatusUpdate { status }, ServerToClientMessage::StatusUpdate { status },
&pool, &ro_pool,
&sockets, &sockets,
) )
.await; .await;
@@ -71,7 +71,7 @@ pub async fn handle_pubsub(
let _ = broadcast_to_local_friends( let _ = broadcast_to_local_friends(
user, user,
ServerToClientMessage::UserOffline { id: user }, ServerToClientMessage::UserOffline { id: user },
&pool, &ro_pool,
&sockets, &sockets,
) )
.await; .await;
@@ -94,7 +94,8 @@ pub async fn handle_pubsub(
notification_id, notification_id,
}) => { }) => {
if let Ok(Some(notification)) = if let Ok(Some(notification)) =
DBNotification::get(notification_id.into(), &pool).await DBNotification::get(notification_id.into(), &*ro_pool)
.await
{ {
let _ = send_notification_to_user( let _ = send_notification_to_user(
&sockets, &sockets,