/** * Opens the WebSocket that `WSProvider` keeps for each tab, and handles what * the client sends over it. Voting, comment likes and every chat write go * through here, not through HTTP. The messages are typed in `shared/ws.ts`, * and `services/ws-service.ts` sends the broadcasts. * * - `GET /ws?token=`, public. A valid session `token` signs the socket in, and * an invalid one leaves it anonymous. Answers 403 to an origin not in * {@link ALLOWED_ORIGINS}, and 426 to a request that isn't an upgrade. * * @module */ import { Router } from "@oak/oak"; import { ALLOWED_ORIGINS } from "../config.ts"; import { authenticate } from "../lib/auth.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 { chatMessageMentionsEveryone, createChatMessage, deleteChatMessage, getChatReadState, markChatRead, updateChatMessage, } from "../services/chat-service.ts"; import { getUnreadCount, notifyEveryone, notifyMentions, } from "../services/notification-service.ts"; import { hasPermission } from "../../shared/permissions.ts"; import { castVote, getUserVotes, removeVote, } from "../services/vote-service.ts"; import { castCommentLike, getUserCommentLikes, removeCommentLike, } from "../services/comment-like-service.ts"; import { getUserById } from "../services/user-service.ts"; import { APIException } from "../lib/api-exception.ts"; import type { ClientToServerMessage, ServerToClientMessage, } from "../../shared/ws.ts"; import type { User } from "../model/user.ts"; /** * Mounted by `api/main.ts` behind the error, Open Graph and CORS middleware. * An `APIException` thrown by a handler becomes the JSON error response there. */ const router: Router = new Router(); /** * Checks whether a socket may be opened from `origin`. Browsers don't apply * CORS to WebSockets, so this is what stops another site from opening one with * its visitors' tokens. */ 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; } // A revoked token (deleted user, changed password) opens an anonymous // socket. The client signs out on its next API call, which gets a 401. const token = ctx.request.url.searchParams.get("token"); const authPayload = token ? await authenticate(token) : null; let user: User | undefined; try { user = authPayload ? getUserById(authPayload.userId) : undefined; } catch { user = undefined; } const socket = ctx.upgrade(); const client: WsClient = { socket, userId: user?.id, username: user?.username, role: user?.role, avatarMime: user?.avatarMime, }; // addEventListener, since `onopen =` proved unreliable under Deno.serve. socket.addEventListener("open", () => { register(client); // The new socket is registered, so it gets this presence_update too, // before its welcome. broadcastPresence(); try { const myVotes = user ? getUserVotes(user.id) : []; const myCommentLikes = user ? getUserCommentLikes(user.id) : []; const unreadNotificationCount = user ? getUnreadCount(user.id) : 0; const chatRead = user ? getChatReadState(user.id) : 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) => { // An exception escaping a WebSocket listener is uncaught in Deno and ends // the process, so nothing a client sends may throw out of here. try { handleMessage(client, event.data); } catch (err) { console.error("[ws] message handler failed:", err); } }); socket.addEventListener("close", () => { unregister(client); broadcastPresence(); // The sweep only sees registered clients, so it would never clear this one. if (client.typingAt) broadcastTyping(); }); }); /** * Parses one message from a client and dispatches it by `type`. A message that * isn't JSON, or whose `type` is unknown, is ignored. Its fields are still * untrusted, so each handler checks their types. */ function handleMessage(client: WsClient, data: unknown): void { if (typeof data !== "string") return; let parsed: unknown; try { parsed = JSON.parse(data); } catch { return; } if ( typeof parsed !== "object" || parsed === null || typeof (parsed as { type?: unknown }).type !== "string" ) return; const msg = parsed as ClientToServerMessage; const { socket } = client; switch (msg.type) { case "ping": send(socket, { 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 dropped, with no error. 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; } } /** Sends `data` unless the socket is closing, where `send` would throw. */ function send(socket: WebSocket, data: ServerToClientMessage): void { if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify(data)); } /** * Handles `vote_cast` and `vote_remove`. It acks the voter with `vote_ack` and * broadcasts `votes_update`. A failure goes back as an `error` naming the dump * and the action, so the client can undo its optimistic vote at once. */ function handleVote( client: WsClient, dumpId: unknown, action: "cast" | "remove", ): void { const { socket } = client; if (!client.userId) { send(socket, { type: "error", message: "Authentication required" }); return; } if (typeof dumpId !== "string") return; try { const { voteCount, isPrivate } = action === "cast" ? castVote(dumpId, client.userId) : removeVote(dumpId, client.userId); send(socket, { type: "vote_ack", dumpId, action, voteCount, }); // Only a private dump's owner can see it, and so vote on it. broadcastVoteUpdate( dumpId, voteCount, client.userId, action, isPrivate ? client.userId : undefined, ); } catch (err) { const message = err instanceof APIException ? err.message : "Vote failed"; send(socket, { type: "error", message, dumpId, action }); } } /** * Handles `comment_like_cast` and `comment_like_remove`, as * {@link handleVote} does votes. */ function handleCommentLike( client: WsClient, commentId: unknown, action: "cast" | "remove", ): void { const { socket } = client; if (!client.userId) { send(socket, { type: "error", message: "Authentication required" }); return; } if (typeof commentId !== "string") return; try { const { likeCount, isPrivate } = action === "cast" ? castCommentLike(commentId, client.userId) : removeCommentLike(commentId, client.userId); send(socket, { type: "comment_like_ack", commentId, action, likeCount, }); broadcastCommentLikeUpdate( commentId, likeCount, client.userId, action, isPrivate ? client.userId : undefined, ); } catch (err) { const message = err instanceof APIException ? err.message : "Like failed"; send(socket, { type: "error", message, commentId, action }); } } /** * Limits each user's chat writes, sends, edits and deletes together, to a burst * of 8, then one every 1.5 s. The buckets are per user, so more tabs don't buy * more. */ const CHAT_WRITE_LIMIT: RateLimit = { capacity: 8, refill: 1 / 1.5 }; const TYPING_LIMIT: RateLimit = { capacity: 10, refill: 1 }; /** * Limits each user's `chat_read`. One over the limit is dropped, and the next * one catches up, since a marker only moves forward. */ const CHAT_READ_LIMIT: RateLimit = { capacity: 10, refill: 1 }; /** * Sends a chat-scoped `error`. With the `nonce` of a `chat_send`, the chatbox * marks that pending message failed. Without one, it shows `message` as a * notice. */ function sendChatError( client: WsClient, message: string, nonce?: string, ): void { send(client.socket, { type: "error", scope: "chat", message, nonce }); } /** * Returns the id of the user behind a chat write, or undefined after sending * the error when they are signed out or over {@link CHAT_WRITE_LIMIT}. */ 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; } /** * Returns `value` when it is a string of at most `maxLength`, else undefined. */ function optionalString(value: unknown, maxLength: number): string | undefined { return typeof value === "string" && value.length <= maxLength ? value : undefined; } /** * Handles `chat_send`. It stores the message, broadcasts it, and notifies the * @mentioned users who aren't reading the chat, and everyone who isn't when * the message is flagged `mentionsEveryone`. */ 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), ); // Before the broadcast, so the typing indicator doesn't linger beside the // message. setClientTyping(client, false); broadcastChatMessage(message, nonce); notifyMentions( userId, message.body, "chat", message.id, "", undefined, isUserChatOpen, ); if (message.mentionsEveryone) { notifyEveryone(userId, message.id, isUserChatOpen); } } catch (err) { sendChatError( client, err instanceof APIException ? err.message : "Failed to send message", nonce, ); } } /** * Checks whether the client's user may edit and delete others' chat messages. * The role is read from the database, not from the socket's connect, so that a * demoted moderator loses the right at once. */ 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"); } /** * Handles `chat_edit`. It stores the new body, broadcasts it, and notifies, in * the author's name, the users the edit newly mentions, and everyone when the * edit adds a `@everyone` the author may say. */ 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 { // Only an edit that adds `@everyone` notifies everyone: fixing a typo in // an announcement mustn't reach those who had the chat open when it was // sent, nor those who joined since. const wasEveryone = chatMessageMentionsEveryone(id); const message = updateChatMessage( id, body, userId, canModerateChat(client), ); broadcastChatMessageUpdated(message); // Mentions already notified for this message are deduplicated, so only // the new ones notify. A moderator's edit keeps the author's words, so the // author mentions. notifyMentions( message.userId, message.body, "chat", message.id, "", undefined, isUserChatOpen, ); if (message.mentionsEveryone && !wasEveryone) { notifyEveryone(message.userId, message.id, isUserChatOpen); } } catch (err) { sendChatError( client, err instanceof APIException ? err.message : "Failed to edit message", ); } } /** Handles `chat_delete`. It deletes the message and broadcasts its id. */ 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", ); } } /** * Handles `chat_read`. It moves the user's read marker and sends * `chat_read_update` to all their sockets, so every tab and device updates 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, and the client doesn't wait for an // answer. } } export default router;