import { Router } from "@oak/oak"; import { ALLOWED_ORIGINS } from "../config.ts"; import { verifyJWT } from "../lib/jwt.ts"; import { broadcastChatMessage, broadcastChatMessageDeleted, broadcastChatMessageUpdated, broadcastCommentLikeUpdate, broadcastPresence, broadcastTyping, broadcastVoteUpdate, getOnlineUsers, handleClientPong, isUserChatOpen, type RateLimit, register, sendToUser, setClientTyping, takeToken, unregister, type WsClient, } from "../services/ws-service.ts"; import { createChatMessage, deleteChatMessage, getChatReadState, markChatRead, updateChatMessage, } from "../services/chat-service.ts"; import { notifyMentions } from "../services/notification-service.ts"; import { hasPermission } from "../lib/permissions.ts"; import { castVote, getUserVotes, removeVote, } from "../services/vote-service.ts"; import { castCommentLike, getUserCommentLikes, removeCommentLike, } from "../services/comment-like-service.ts"; import { getUnreadCount } from "../services/notification-service.ts"; import { getUserById } from "../services/user-service.ts"; import { APIException, type ClientToServerMessage, } from "../model/interfaces.ts"; const router = new Router(); function isAllowedOrigin(origin: string): boolean { return ALLOWED_ORIGINS.includes(origin); } router.get("/ws", async (ctx) => { const origin = ctx.request.headers.get("origin") ?? ""; if (!isAllowedOrigin(origin)) { ctx.response.status = 403; return; } if (!ctx.isUpgradable) { ctx.response.status = 426; return; } const token = ctx.request.url.searchParams.get("token"); const authPayload = token ? await verifyJWT(token) : null; const socket = ctx.upgrade(); // Read the live user so the role reflects the DB, not the (possibly stale) // role embedded in the 7-day JWT. const user = authPayload ? getUserById(authPayload.userId) : undefined; const client: WsClient = { socket, userId: authPayload?.userId, username: authPayload?.username, role: user?.role, avatarMime: user?.avatarMime, }; // Use addEventListener — more reliable than onopen= with Deno.serve socket.addEventListener("open", () => { register(client); broadcastPresence(); try { const myVotes = authPayload ? getUserVotes(authPayload.userId) : []; const myCommentLikes = authPayload ? getUserCommentLikes(authPayload.userId) : []; const unreadNotificationCount = authPayload ? getUnreadCount(authPayload.userId) : 0; const chatRead = authPayload ? getChatReadState(authPayload.userId) : undefined; socket.send(JSON.stringify({ type: "welcome", users: getOnlineUsers(), myVotes, myCommentLikes, unreadNotificationCount, chatReadAt: chatRead?.readAt, unreadChatCount: chatRead?.unread ?? 0, })); } catch (err) { console.error("[ws] welcome send failed:", err); } }); socket.addEventListener("message", (event) => { let msg: ClientToServerMessage; try { msg = JSON.parse(event.data as string); } catch { return; } switch (msg.type) { case "ping": socket.send(JSON.stringify({ type: "pong" })); break; case "pong": handleClientPong(client); break; case "vote_cast": handleVote(client, msg.dumpId, "cast"); break; case "vote_remove": handleVote(client, msg.dumpId, "remove"); break; case "comment_like_cast": handleCommentLike(client, msg.commentId, "cast"); break; case "comment_like_remove": handleCommentLike(client, msg.commentId, "remove"); break; case "chat_send": handleChatSend(client, msg.body, msg.replyToId, msg.nonce); break; case "chat_typing": // Typing is cosmetic: past the limit it's just dropped. if ( client.userId && takeToken(`typing:${client.userId}`, TYPING_LIMIT) ) { setClientTyping(client, msg.typing === true); } break; case "chat_focus": client.chatOpen = msg.open === true; break; case "chat_edit": handleChatEdit(client, msg.id, msg.body); break; case "chat_delete": handleChatDelete(client, msg.id); break; case "chat_read": handleChatRead(client, msg.at); break; } }); socket.addEventListener("close", () => { unregister(client); broadcastPresence(); // If they were mid-compose, drop them from the typing set right away. if (client.typingAt) broadcastTyping(); }); }); function handleVote( client: WsClient, dumpId: string, action: "cast" | "remove", ): void { const { socket } = client; if (!client.userId) { socket.send( JSON.stringify({ type: "error", message: "Authentication required" }), ); return; } try { const newCount = action === "cast" ? castVote(dumpId, client.userId) : removeVote(dumpId, client.userId); socket.send(JSON.stringify({ type: "vote_ack", dumpId, action, success: true, voteCount: newCount, })); broadcastVoteUpdate(dumpId, newCount, client.userId, action); } catch (err) { const message = err instanceof APIException ? err.message : "Vote failed"; socket.send(JSON.stringify({ type: "error", message, dumpId, action })); } } function handleCommentLike( client: WsClient, commentId: string, action: "cast" | "remove", ): void { const { socket } = client; if (!client.userId) { socket.send( JSON.stringify({ type: "error", message: "Authentication required" }), ); return; } try { const newCount = action === "cast" ? castCommentLike(commentId, client.userId) : removeCommentLike(commentId, client.userId); socket.send(JSON.stringify({ type: "comment_like_ack", commentId, action, likeCount: newCount, })); broadcastCommentLikeUpdate(commentId, newCount, client.userId, action); } catch (err) { const message = err instanceof APIException ? err.message : "Like failed"; socket.send(JSON.stringify({ type: "error", message, commentId, action })); } } // Sending, editing and deleting share one bucket: a burst of 8, then one every // 1.5s — room for a quick back-and-forth, not for a flood. const CHAT_WRITE_LIMIT: RateLimit = { capacity: 8, refill: 1 / 1.5 }; const TYPING_LIMIT: RateLimit = { capacity: 10, refill: 1 }; // Read markers are cheap for the client to send often; past this they're // dropped, and the next one catches up. const CHAT_READ_LIMIT: RateLimit = { capacity: 10, refill: 1 }; function sendChatError( client: WsClient, message: string, nonce?: string, ): void { if (client.socket.readyState !== WebSocket.OPEN) return; client.socket.send( JSON.stringify({ type: "error", scope: "chat", message, nonce }), ); } // Validates the common preconditions of a chat write: signed in and under the // rate limit. Returns the user id, or undefined after reporting why not. function chatWriter(client: WsClient, nonce?: string): string | undefined { if (!client.userId) { sendChatError(client, "Authentication required", nonce); return undefined; } if (!takeToken(`chat:${client.userId}`, CHAT_WRITE_LIMIT)) { sendChatError( client, "You're sending messages too fast — wait a moment", nonce, ); return undefined; } return client.userId; } function optionalString(value: unknown, maxLength: number): string | undefined { return typeof value === "string" && value.length <= maxLength ? value : undefined; } function handleChatSend( client: WsClient, body: unknown, replyToId: unknown, rawNonce: unknown, ): void { const nonce = optionalString(rawNonce, 64); const userId = chatWriter(client, nonce); if (!userId) return; if (typeof body !== "string") { sendChatError(client, "Message cannot be empty", nonce); return; } try { const message = createChatMessage( userId, body, optionalString(replyToId, 64), ); // Sending a message ends composition — clear the typing flag before the // message broadcast so the indicator doesn't linger next to it. setClientTyping(client, false); // Broadcast to everyone including the sender, whose pending copy the nonce // retires. broadcastChatMessage(message, nonce); // Skip mention notifications for users who already have the chat open — // they've seen the message live. notifyMentions( userId, message.body, "chat", message.id, "", undefined, isUserChatOpen, ); } catch (err) { sendChatError( client, err instanceof APIException ? err.message : "Failed to send message", nonce, ); } } // Reads the role from the DB on every moderation check rather than trusting the // one captured at connect: a demoted moderator loses the power right away, not // whenever their socket next reconnects. function canModerateChat(client: WsClient): boolean { if (!client.userId) return false; try { client.role = getUserById(client.userId).role; } catch { client.role = undefined; } return !!client.role && hasPermission(client.role, "chat:moderate"); } function handleChatEdit(client: WsClient, id: unknown, body: unknown): void { const userId = chatWriter(client); if (!userId) return; if (typeof id !== "string" || typeof body !== "string") { sendChatError(client, "Invalid message"); return; } try { const message = updateChatMessage( id, body, userId, canModerateChat(client), ); broadcastChatMessageUpdated(message); // A newly added @mention in an edit should still notify (deduped per // message+user), skipping anyone already reading the chat. notifyMentions( userId, message.body, "chat", message.id, "", undefined, isUserChatOpen, ); } catch (err) { sendChatError( client, err instanceof APIException ? err.message : "Failed to edit message", ); } } function handleChatDelete(client: WsClient, id: unknown): void { const userId = chatWriter(client); if (!userId) return; if (typeof id !== "string") { sendChatError(client, "Invalid message"); return; } try { deleteChatMessage(id, userId, canModerateChat(client)); broadcastChatMessageDeleted(id); } catch (err) { sendChatError( client, err instanceof APIException ? err.message : "Failed to delete message", ); } } // Moves the user's read marker and tells all their sockets, so every tab and // device drops its badge together. function handleChatRead(client: WsClient, at: unknown): void { if (!client.userId || typeof at !== "string") return; if (!takeToken(`read:${client.userId}`, CHAT_READ_LIMIT)) return; try { const state = markChatRead(client.userId, at); sendToUser(client.userId, { type: "chat_read_update", readAt: state.readAt, unread: state.unread, }); } catch { // A malformed marker changes nothing; there's nothing to report. } } export default router;