Overhaul watch together system

src/events/ban.ts

@@ -1,70 +1,67 @@
 import type { Server, Socket } from 'socket.io';
 
-import { get, set } from '@utils/cache';
 import canDoModerationOperationOnTarget from '@utils/canDoModerationOperationOnTarget';
 import sendSystemMessage from '@utils/systemMessage';
+import useSocket from '@utils/useSocket';
 
-import { CoreParticipant, Participant } from './login';
+import type { CoreParticipant } from './create';
 
 export default class BanOrUnbanParticipant {
     async handle({ socket, io, data }: { socket: Socket; io: Server; data: any }) {
         const targetUserId = data?.target;
         if (!targetUserId) return;
 
-        const rooms = Array.from(socket.rooms);
-        const room = rooms[1];
+        const hook = useSocket(socket);
+        if (hook?.error) return;
 
-        if (!room) return;
+        const room = hook.getRoomPtr();
+        const mod = await hook.getCurrentUser();
+        const targetUser = await hook.getUserFromId(targetUserId);
+        const bannedParticipants = await hook.getBannedParticipants();
 
-        const socketId = socket.id;
-        const socketRoomParticipants = (await get(`room:${room}:users`)) as Participant[];
+        let newBannedParticipants: CoreParticipant[] = bannedParticipants;
+        let shouldBroadcast = false;
 
-        if (socketRoomParticipants) {
-            const mod = socketRoomParticipants.find((user) => user.sid == socketId);
-            const targetUser = socketRoomParticipants.find((user) => user.id == targetUserId);
+        const isBanned = bannedParticipants.find((x) => x.id == targetUserId);
 
-            const bannedParticipants = ((await get(`room:${room}:bannedParticipants`)) ??
-                []) as CoreParticipant[];
+        if (isBanned) {
+            // unban
+            if (mod?.moderator) {
+                newBannedParticipants = bannedParticipants.filter((x) => x.id != targetUserId);
+                shouldBroadcast = true;
 
-            let newBannedParticipants = [];
-
-            const isBanned = bannedParticipants.find((x) => x.id == targetUserId);
-
-            if (isBanned) {
-                // unban
-                if (mod?.moderator) {
-                    newBannedParticipants = bannedParticipants.filter((x) => x.id != targetUserId);
-
-                    sendSystemMessage(
-                        room,
-                        `${mod.username}, ${isBanned.username} kullanıcısının yasağını kaldırdı.`,
-                    );
-                }
-            } else if (mod && targetUser && canDoModerationOperationOnTarget(mod, targetUser)) {
-                //ban
+                sendSystemMessage(
+                    room,
+                    `${mod.username}, ${isBanned.username} kullanıcısının yasağını kaldırdı.`,
+                );
+            }
+        } else if (mod && targetUser && canDoModerationOperationOnTarget(mod, targetUser)) {
+            //ban
 
-                const targetUserCpy = { ...targetUser };
+            const targetUserCpy = { ...targetUser };
 
-                delete targetUserCpy.sid;
-                delete targetUserCpy.owner;
-                delete targetUserCpy.moderator;
+            delete targetUserCpy.sid;
+            delete targetUserCpy.owner;
+            delete targetUserCpy.moderator;
 
-                newBannedParticipants = [...bannedParticipants, targetUserCpy];
+            newBannedParticipants = [...bannedParticipants, targetUserCpy];
+            shouldBroadcast = true;
 
-                const getTargetSocket = io.sockets.sockets.get(targetUser.sid);
-                getTargetSocket.disconnect();
+            const getTargetSocket = io.sockets.sockets.get(targetUser.sid);
+            getTargetSocket?.disconnect();
 
-                sendSystemMessage(
-                    room,
-                    `${mod.username}, ${targetUser.username} kullanıcısını yasakladı.`,
-                );
-            }
+            sendSystemMessage(
+                room,
+                `${mod.username}, ${targetUser.username} kullanıcısını yasakladı.`,
+            );
+        }
 
-            io.in(room).emit('ban', {
+        if (shouldBroadcast) {
+            hook.broadcastToEveryone('ban', {
                 bannedParticipants: newBannedParticipants,
             });
 
-            await set(`room:${room}:bannedParticipants`, newBannedParticipants);
+            await hook.setRoomKey('bannedParticipants', newBannedParticipants);
         }
     }
 }

src/events/create.ts

@@ -0,0 +1,161 @@
+import crypto from 'node:crypto';
+
+import { Socket } from 'socket.io';
+import { z } from 'zod';
+
+import { chatBotProps, io } from '@index';
+import { get, multipleSet, set } from '@utils/cache';
+import sendSystemMessage from '@utils/systemMessage';
+import { addTurkishPossessiveSuffix } from '@utils/turkishPossessiveSuffix';
+import useSocket from '@utils/useSocket';
+
+export type Participant = {
+    id: string;
+    username: string;
+    avatar: string;
+    avatarDecoration: number;
+    owner: boolean;
+    moderator: boolean;
+    sid: string;
+};
+
+export type CoreParticipant = Omit<Participant, 'sid' | 'owner' | 'moderator'>;
+
+export type SocketSession = {
+    room: string;
+    id: string;
+    participant?: Participant;
+};
+
+export async function getParticipantsFromSocketRoom(room: string) {
+    const roomSockets = io.sockets.adapter.rooms.get(room);
+    if (!roomSockets) return [];
+
+    const participants = await Promise.all(
+        [...roomSockets].map(async (sid) => {
+            const session = (await get(`sid:${sid}`)) as SocketSession | null;
+
+            if (session?.room != room) return null;
+            return session.participant ?? null;
+        }),
+    );
+
+    return participants.filter((participant): participant is Participant => !!participant);
+}
+
+const validation = z.object({
+    token: z.string().max(1000),
+    anime: z.object({
+        fansub: z.string().min(1).max(500),
+        slug: z.string().min(1).max(500),
+        season: z.number().int(),
+        episode: z.number().int(),
+    }),
+    timestamp: z.number().nonnegative().optional(),
+});
+
+export default class CreateRoom {
+    async handle({ socket, callback, data }: { socket: Socket; callback: any; data: any }) {
+        const token = data?.token || socket.handshake.headers.authorization;
+
+        const val = validation.safeParse({
+            ...data,
+            token,
+        });
+
+        if ('error' in val) {
+            let err = val.error.issues[0].message;
+
+            if (err == 'Required') err = 'Invalid body';
+            return callback({ error: err });
+        }
+
+        const { anime, timestamp } = data;
+
+        const user = (await fetch(`${process.env.API_URL}/user`, {
+            headers: {
+                Authorization: token,
+                'Client-Protocol-Model': process.env.CLIENT_PROTOCOL_MODEL_VALUE,
+            },
+        })) as any;
+
+        // allocate 4 bytes. in hex, a byte is represented by 2 chars so its n * 2 - so 8 chars.
+        const roomId = 'room:' + crypto.randomBytes(4).toString('hex');
+
+        const json = await user.json();
+        if (!json?.id) return callback({ error: 'Kullanıcı verisi alınamadı' });
+
+        let roomParticipants = ((await get(`${roomId}:users`)) ?? []) as Participant[];
+        const participantsDefinedBySocketIO = io.sockets.adapter.rooms.get(roomId);
+
+        if (participantsDefinedBySocketIO) {
+            roomParticipants = roomParticipants.filter((participant) =>
+                participantsDefinedBySocketIO.has(participant.sid),
+            );
+
+            if (roomParticipants.length == 0) {
+                roomParticipants = await getParticipantsFromSocketRoom(roomId);
+            }
+        }
+
+        if (roomParticipants && roomParticipants.find((user) => user.id == json.id)) {
+            return callback({ error: 'Zaten bu odadasın' });
+        }
+
+        const roomName = `${addTurkishPossessiveSuffix(json.username)} odası`;
+
+        const currentParticipant: Participant = {
+            id: json.id,
+            username: json.username,
+            avatar: json.avatar,
+            avatarDecoration: json?.avatarDecoration ?? 0,
+            owner: true,
+            moderator: true,
+            sid: socket.id,
+        };
+
+        await multipleSet({
+            [`${roomId}:users`]: [currentParticipant],
+            [`${roomId}:timestamp`]: timestamp ?? 0,
+            [`${roomId}:anime`]: anime,
+            [`${roomId}:owner`]: json.id,
+            [`${roomId}:password`]: null,
+            [`${roomId}:bannedParticipants`]: [],
+            [`${roomId}:mutedParticipants`]: [],
+            [`${roomId}:controlledByMods`]: false,
+            [`${roomId}:name`]: roomName,
+        });
+
+        await set(`sid:${socket.id}`, {
+            room: roomId,
+            id: json.id,
+            participant: currentParticipant,
+        });
+
+        socket.join(roomId);
+
+        sendSystemMessage(roomId, `${json.username} odaya katıldı 👋`);
+
+        setTimeout(async () => {
+            const hook = useSocket(socket);
+            if (hook?.error) return;
+
+            hook.broadcastToEveryone('participants', {
+                participants: await hook.getParticipants(),
+            });
+        }, 1000);
+
+        return callback({
+            message: 'OK',
+            details: {
+                bannedParticipants: [],
+                mutedParticipants: [],
+                timestamp: timestamp ?? 0,
+                roomId: roomId,
+                roomName: roomName,
+                controlledByMods: false,
+            },
+            system: chatBotProps,
+        });
+    }
+}

src/events/disconnecting.ts

@@ -1,31 +1,77 @@
 import type { Server, Socket } from 'socket.io';
 
-import { del, delWithPattern, get, set } from '@utils/cache';
+import type { Participant } from '@events/login';
+
+import { del, delWithPattern, get } from '@utils/cache';
+import useSocket from '@utils/useSocket';
+
+type SocketSession = {
+    room: string;
+    id: string;
+    participant?: Participant;
+};
+
+async function getParticipantsFromSocketIds(room: string, socketIds: string[]) {
+    const participants = await Promise.all(
+        socketIds.map(async (sid) => {
+            const session = (await get(`sid:${sid}`)) as SocketSession | null;
+
+            if (session?.room != room) return null;
+            return session.participant ?? null;
+        }),
+    );
+
+    return participants.filter((participant): participant is Participant => !!participant);
+}
 
 export default class Disconnect {
     async handle({ socket, io }: { socket: Socket; io: Server }) {
-        const rooms = Array.from(socket.rooms);
-        const room = rooms[1];
+        const hook = useSocket(socket);
+        if (hook?.error) return;
 
-        if (!room) return;
+        const room = hook.getRoomPtr();
+        const socketRoomParticipants = await hook.getParticipants();
+        const remainingSocketIds = [...(io.sockets.adapter.rooms.get(room) ?? [])].filter(
+            (sid) => sid != socket.id,
+        );
 
-        const socketId = socket.id;
-        const socketRoomParticipants = await get(`room:${room}:users`);
+        await del(`sid:${socket.id}`);
 
         if (socketRoomParticipants) {
-            const newParticipants = socketRoomParticipants.filter((user) => user.sid != socketId);
+            const newParticipants = socketRoomParticipants.filter((user) => user.sid != socket.id);
+
+            if (newParticipants.length == 0) {
+                const rebuiltParticipants = await getParticipantsFromSocketIds(
+                    room,
+                    remainingSocketIds,
+                );
 
-            await set(`room:${room}:users`, newParticipants);
+                if (rebuiltParticipants.length > 0) {
+                    await hook.setRoomKey('users', rebuiltParticipants);
 
-            io.in(room).emit('participants', {
+                    hook.broadcastToEveryone('participants', {
+                        participants: rebuiltParticipants,
+                    });
+
+                    return;
+                }
+
+                if (remainingSocketIds.length == 0) {
+                    await delWithPattern(`room:${room}:*`);
+                }
+
+                return;
+            }
+
+            await hook.setRoomKey('users', newParticipants);
+
+            hook.broadcastToEveryone('participants', {
                 participants: newParticipants,
             });
-        }
 
-        const remainingParticipants = io.sockets.adapter.rooms.get(room);
-        if (!remainingParticipants) {
-            await delWithPattern(`room:${room}:*`);
-            await del(`sid:${socketId}`);
+            return;
         }
+
+        if (remainingSocketIds.length == 0) await delWithPattern(`room:${room}:*`);
     }
 }