Ryan/drop
1
0
Fork 0
mirror of https://github.com/Drop-OSS/drop.git synced 2025-11-09 20:12:10 +10:00
Code Issues Packages Projects Releases Wiki Activity

fix: convert socket sessions to cacheHandler

Browse Source
This commit is contained in:
DecDuck
2025-05-08 15:50:29 +10:00
parent e3ed60feae
commit 733aee3977
2 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
server/api/v1/notifications/ws.get.ts
View File

@ -1,9 +1,9 @@
import notificationSystem from "~/server/internal/notifications";
import aclManager from "~/server/internal/acls";
import cacheHandler from "~/server/internal/cache";
// TODO add web socket sessions for horizontal scaling
// Peer ID to user ID
const socketSessions = new Map<string, string>();
const socketSessions = cacheHandler.createCache<string>("notificationSocketSessions");
export default defineWebSocketHandler({
async open(peer) {
@ -23,7 +23,7 @@ export default defineWebSocketHandler({
userIds.push("system");
}
socketSessions.set(peer.id, userId);
await socketSessions.set(peer.id, userId);
for (const listenUserId of userIds) {
notificationSystem.listen(listenUserId, peer.id, (notification) => {
@ -32,7 +32,7 @@ export default defineWebSocketHandler({
}
},
async close(peer, _details) {
const userId = socketSessions.get(peer.id);
const userId = await socketSessions.get(peer.id);
if (!userId) {
console.log(`skipping websocket close for ${peer.id}`);
return;
@ -40,6 +40,6 @@ export default defineWebSocketHandler({
notificationSystem.unlisten(userId, peer.id);
notificationSystem.unlisten("system", peer.id); // In case we were listening as 'system'
socketSessions.delete(peer.id);
await socketSessions.remove(peer.id);
},
});

16
server/api/v1/task/index.get.ts
View File

@ -1,9 +1,9 @@
import taskHandler from "~/server/internal/tasks";
import type { MinimumRequestObject } from "~/server/h3";
import cacheHandler from "~/server/internal/cache";
// TODO add web socket sessions for horizontal scaling
// ID to admin
const socketHeaders = new Map<string, MinimumRequestObject>();
const socketHeaders = cacheHandler.createCache<MinimumRequestObject>("taskSocketHeaders");
export default defineWebSocketHandler({
async open(peer) {
@ -13,15 +13,15 @@ export default defineWebSocketHandler({
return;
}
socketHeaders.set(peer.id, {
await socketHeaders.set(peer.id, {
headers: request.headers ?? new Headers(),
});
peer.send(`connect`);
},
message(peer, message) {
async message(peer, message) {
if (!peer.id) return;
const headers = socketHeaders.get(peer.id);
if (headers === undefined) return;
const headers = await socketHeaders.get(peer.id);
if (!headers) return;
const text = message.text();
if (text.startsWith("connect/")) {
const id = text.substring("connect/".length);
@ -29,10 +29,10 @@ export default defineWebSocketHandler({
return;
}
},
close(peer, _details) {
async close(peer, _details) {
if (!peer.id) return;
if (!socketHeaders.has(peer.id)) return;
socketHeaders.delete(peer.id);
await socketHeaders.remove(peer.id);
taskHandler.disconnectAll(peer.id);
},