API core

backend/src/constants.ts

@@ -0,0 +1,4 @@
+export const REAPER_INTERVAL_MS = 15_000
+export const REAPER_TIMEOUT_MS = 30_000
+
+export const ROOM_MAX_CHAT_HISTORY = 50

backend/src/handlers/common.ts

@@ -0,0 +1,80 @@
+import type { ClientMessages, UserMessage } from '@sync/wire'
+import type { HandlerMap } from '.'
+import type { SyncServer, SyncWS } from '../server'
+import { serializeMsg } from '@sync/wire/backend'
+import { MediaManager } from '../services/MediaManager'
+import { notOwnerGuard } from '../util/guards'
+import { reply, replyError } from '../util/msg'
+
+export const COMMON_HANDLERS = {
+    ping: (ws: SyncWS, msg: ClientMessages['ping']) => {
+        reply(ws, msg, 'pong', {
+            timestamp: Date.now(),
+        })
+    },
+
+    struggle: (ws: SyncWS, msg: ClientMessages['struggle']) => {
+        const me = ws.data.user
+        const owner = me.room.owner
+        if (!owner) return
+
+        if (notOwnerGuard(ws, msg)) return
+
+        // TODO: Rate limit.
+
+        ws.data.user.room.owner?.webSocket?.send(
+            serializeMsg('userStruggle', { userId: ws.data.user.id }),
+        )
+    },
+
+    message: async (ws: SyncWS, msg: ClientMessages['message'], server: SyncServer) => {
+        let text = msg.body.text?.trim()
+        if (text?.length === 0) text = undefined
+
+        //TODO: Rate limit
+
+        if (!text && !msg.body.recommendation) {
+            replyError(ws, msg, {
+                type: 'invalidChatMessage',
+                message: 'Message must contain text or a media recommendation.',
+            })
+            return
+        }
+
+        const recommendation = msg.body.recommendation
+
+        if (recommendation && !(await MediaManager.verifyMedia(recommendation))) {
+            replyError(ws, msg, {
+                type: 'invalidMedia',
+                message: 'The recommended media is invalid.',
+            })
+            return
+        }
+
+        const cmsg: UserMessage = {
+            type: 'user',
+            userId: ws.data.user.id,
+            timestamp: Date.now(),
+            text,
+            recommendation: msg.body.recommendation,
+        }
+
+        ws.data.user.room.addMessage(cmsg)
+        server.publish(ws.data.user.room.topic, serializeMsg('chatMessage', cmsg))
+    },
+
+    playbackStats: (ws: SyncWS, msg: ClientMessages['playbackStats']) => {
+        const me = ws.data.user
+        const owner = me.room.owner
+
+        if (!owner) return
+
+        if (notOwnerGuard(ws, msg)) return
+
+        // TODO: Rate limit.
+
+        ws.data.user.room.owner?.webSocket?.send(
+            serializeMsg('playbackReport', { userId: ws.data.user.id, stats: msg.body }),
+        )
+    },
+} satisfies Partial<HandlerMap>

backend/src/handlers/index.ts

@@ -0,0 +1,17 @@
+import type { ClientMessage } from '@sync/wire'
+import type { SyncServer, SyncWS } from '../server'
+import { OWNER_HANDLERS } from './owner'
+import { COMMON_HANDLERS } from './common'
+
+export type HandlerMap = {
+    [TKey in ClientMessage['type']]: (
+        ws: SyncWS,
+        msg: Extract<ClientMessage, { type: TKey }>,
+        server: SyncServer,
+    ) => void | Promise<void>
+}
+
+export const HANDLERS = {
+    ...OWNER_HANDLERS,
+    ...COMMON_HANDLERS,
+} satisfies HandlerMap

backend/src/handlers/owner.ts

@@ -0,0 +1,166 @@
+import type { ClientMessages } from '@sync/wire'
+import type { HandlerMap } from '.'
+import type { SyncServer, SyncWS } from '../server'
+import { CloseCode, CloseReason, serializeMsg } from '@sync/wire/backend'
+import { Room } from '../models'
+import { SessionManager } from '../services/SessionManager'
+import { RoomManager } from '../services/RoomManager'
+import { notSelfGuard, ownerGuard, targetUserExistsGuard } from '../util/guards'
+import { broadcast, reply, replyError, replyOk } from '../util/msg'
+
+export const OWNER_HANDLERS = {
+    sync: (ws: SyncWS, msg: ClientMessages['sync'], server: SyncServer) => {
+        if (ownerGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        room.setSync(msg.body)
+
+        broadcast(server, ws, 'ssync', msg.body)
+    },
+
+    updateRoom: (ws: SyncWS, msg: ClientMessages['updateRoom'], server: SyncServer) => {
+        if (ownerGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        let changed = false
+
+        if (msg.body.name) {
+            if (Room.checkName) {
+                const nameError = Room.checkName(msg.body.name)
+
+                if (nameError) {
+                    return replyError(ws, msg, {
+                        type: 'badRoomUpdate',
+                        message: nameError,
+                    })
+                }
+            }
+
+            room.name = msg.body.name.trim()
+            changed = true
+        }
+
+        replyOk(ws, msg)
+
+        if (changed) {
+            broadcast(server, ws, 'roomUpdated', { room: { name: room.name, slug: room.slug } })
+        }
+    },
+
+    clearChat: (ws: SyncWS, msg: ClientMessages['clearChat'], server: SyncServer) => {
+        if (ownerGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        room.chat.length = 0
+
+        broadcast(server, ws, 'chatCleared', null)
+    },
+
+    destroyRoom: (ws: SyncWS, msg: ClientMessages['destroyRoom']) => {
+        if (ownerGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        for (const user of room.users.values()) {
+            if (!user.webSocket) continue
+
+            user.webSocket.data.closedByServer = true
+            user.webSocket.close(CloseCode.RoomClosed, CloseReason.RoomClosed)
+            user.webSocket = undefined
+            SessionManager.destroy(user.sessionId)
+        }
+
+        room.users.clear()
+        RoomManager.deleteRoom(room.slug)
+    },
+
+    promote: (ws: SyncWS, msg: ClientMessages['promote'], server: SyncServer) => {
+        if (ownerGuard(ws, msg)) return
+        if (notSelfGuard(ws, msg)) return
+        if (targetUserExistsGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        const targetUser = room.users.get(msg.body.userId)
+        if (!targetUser) return // TODO: Assert/errror/...
+
+        room.promote(targetUser)
+
+        broadcast(server, ws, 'roomUpdated', { ownerId: targetUser.id })
+    },
+
+    kick: (ws: SyncWS, msg: ClientMessages['kick'], server: SyncServer) => {
+        if (ownerGuard(ws, msg)) return
+        if (notSelfGuard(ws, msg)) return
+        if (targetUserExistsGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        const targetUser = room.users.get(msg.body.userId)
+
+        if (!targetUser) return // TODO: Assert/errror/...
+
+        if (targetUser.webSocket) {
+            targetUser.webSocket.data.closedByServer = true
+            targetUser.webSocket.close(CloseCode.Kicked, CloseReason.Kicked)
+            targetUser.webSocket = undefined
+        }
+
+        room.removeUser(targetUser, () => {})
+        SessionManager.destroy(targetUser.sessionId)
+
+        broadcast(server, ws, 'userLeft', { userId: targetUser.id })
+    },
+
+    kickAll: (ws: SyncWS, msg: ClientMessages['kickAll']) => {
+        if (ownerGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        for (const user of room.users.values()) {
+            if (!user.webSocket || user.webSocket === ws) continue
+
+            user.webSocket.data.closedByServer = true
+            user.webSocket.close(CloseCode.Kicked, CloseReason.Kicked)
+            user.webSocket = undefined
+            SessionManager.destroy(user.sessionId)
+        }
+
+        room.users.clear()
+        room.users.set(ws.data.user.id, ws.data.user)
+
+        reply(ws, msg, 'roomUpdated', { users: [ws.data.user.toWire()] })
+    },
+
+    updatePlaylist: (ws: SyncWS, msg: ClientMessages['updatePlaylist'], server: SyncServer) => {
+        if (ownerGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+
+        const error = room.updatePlaylist(msg.body)
+
+        if (error) {
+            return replyError(ws, msg, {
+                type: 'badPlaylist',
+                message: error,
+            })
+        }
+
+        replyOk(ws, msg)
+        broadcast(server, ws, 'roomUpdated', { playlist: room.playlist })
+    },
+
+    queryPlayback: (ws: SyncWS, msg: ClientMessages['queryPlayback']) => {
+        if (ownerGuard(ws, msg)) return
+        if (notSelfGuard(ws, msg)) return
+        if (targetUserExistsGuard(ws, msg)) return
+
+        const room = ws.data.user.room
+        const targetUser = room.users.get(msg.body.userId)
+
+        targetUser?.webSocket?.send(serializeMsg('playbackQuery', null))
+    },
+} satisfies Partial<HandlerMap>