/** * Keeps the set of open sockets and sends them the server's messages. * * The WebSocket route (`routes/ws.ts`) registers each socket and handles what * it sends. The services and routes that change dumps, comments, playlists and * users call the `broadcast*` functions here after the change is stored. A * `broadcast*` function sends to everyone unless its doc names an audience, so * its caller must not pass it anything private. The module also pings every * socket and closes the ones that stop answering. * * @module */ import type { ChatMessage } from "../../shared/chat.ts"; import type { Comment } from "../../shared/comment.ts"; import type { Dump } from "../../shared/dump.ts"; import type { OnlineUser, ServerToClientMessage } from "../../shared/ws.ts"; import type { Playlist } from "../../shared/playlist.ts"; import type { PublicUser, Role } from "../../shared/user.ts"; import { isMemberVisibleTo, type PlaylistMember, } from "../lib/playlist-members.ts"; /** * Created by the WebSocket route for each socket it opens, and kept here until * the socket closes. The user fields are read at connect and left undefined for * an anonymous socket. */ export interface WsClient { socket: WebSocket; userId?: string; username?: string; /** * Read again from the database on each chat edit or delete, so a demoted * moderator loses the right at once. */ role?: Role; avatarMime?: string; /** Set by {@link updateClientAvatar}, for the client's avatar URLs. */ avatarVersion?: number; /** Undefined until the first ping, so a new socket isn't closed at once. */ pongReceived?: boolean; /** * Set by `chat_focus`: the chatbox is open and the tab visible, so the user * is reading the chat and gets no notification for a mention in it. */ chatOpen?: boolean; /** When the last `chat_typing` true arrived, in ms. */ typingAt?: number; } const clients = new Set(); /** * Adds a client once its socket opens. From then on, it receives the * broadcasts, and is shown online if it is signed in. */ export function register(client: WsClient): void { clients.add(client); } /** * Removes a client when its socket closes. The caller rebroadcasts presence. */ export function unregister(client: WsClient): void { clients.delete(client); } /** * Records `userId`'s new avatar on their clients, so presence and the typing * list show it, and rebroadcasts presence. `routes/avatars.ts` calls it on an * upload. */ export function updateClientAvatar(userId: string, avatarMime: string): void { const version = Date.now(); for (const client of clients) { if (client.userId === userId) { client.avatarMime = avatarMime; client.avatarVersion = version; } } broadcastPresence(); } /** * Returns each signed-in user with a socket open, once. The route sends it in * `welcome`, and {@link broadcastPresence} in `presence_update`. */ export function getOnlineUsers(): OnlineUser[] { const seen = new Map(); for (const client of clients) { if (client.userId && !seen.has(client.userId)) { seen.set(client.userId, { userId: client.userId, username: client.username!, hasAvatar: !!client.avatarMime, avatarVersion: client.avatarVersion, }); } } return Array.from(seen.values()); } /** Sends `payload` unless the socket is closing, where `send` would throw. */ function sendRaw(socket: WebSocket, payload: string): void { if (socket.readyState === WebSocket.OPEN) socket.send(payload); } function send(socket: WebSocket, data: ServerToClientMessage): void { sendRaw(socket, JSON.stringify(data)); } /** * Sends `data` to every client `include` accepts, or to all of them. It is * serialized once, not once per socket. */ function broadcast( data: ServerToClientMessage, include?: (client: WsClient) => boolean, ): void { let payload: string | undefined; for (const client of clients) { if (include && !include(client)) continue; payload ??= JSON.stringify(data); sendRaw(client.socket, payload); } } /** * Sends `data` to every socket of `userId`. The notification service sends * `notification_created` with it, the follow service `follows_changed`, and * the route `chat_read_update`. */ export function sendToUser(userId: string, data: ServerToClientMessage): void { broadcast(data, (client) => client.userId === userId); } /** * Closes every socket of `userId` without signing anyone out. The user service * calls it on a password change. Each tab reconnects with the token it holds: * the tab that made the change has a fresh one and stays signed in, while the * others connect anonymously and sign out on their next 401. */ export function closeUserSockets(userId: string, reason: string): void { for (const client of clients) { if (client.userId === userId) client.socket.close(1000, reason); } } /** * Sends everyone `presence_update`. The route calls it whenever a socket opens * or closes. */ export function broadcastPresence(): void { broadcast({ type: "presence_update", users: getOnlineUsers() }); } /** * Tells the clients that may see a dump how it changed, for the dump service * after every write. `before` is null for a new dump, `after` for a deleted * one. A private dump's changes reach its owner's sockets alone. When a dump * goes private, everyone else gets `dump_deleted`; when it goes public, they * get `dump_created`, since to them it is new. */ export function broadcastDumpChange( before: Dump | null, after: Dump | null, ): void { const dump = after ?? before; if (!dump) return; const ownerId = dump.userId; const toOwner = (client: WsClient) => client.userId === ownerId; const toOthers = (client: WsClient) => client.userId !== ownerId; if (!after) { broadcast( { type: "dump_deleted", dumpId: dump.id }, dump.isPrivate ? toOwner : undefined, ); } else if (!before) { broadcast( { type: "dump_created", dump: after }, after.isPrivate ? toOwner : undefined, ); } else if (after.isPrivate) { broadcast({ type: "dump_updated", dump: after }, toOwner); if (!before.isPrivate) { broadcast({ type: "dump_deleted", dumpId: after.id }, toOthers); } } else if (before.isPrivate) { broadcast({ type: "dump_updated", dump: after }, toOwner); broadcast({ type: "dump_created", dump: after }, toOthers); } else { broadcast({ type: "dump_updated", dump: after }); } } /** * Sends `votes_update` after the route stored a vote, to everyone, or with * `onlyUserId` to that user's sockets alone (for a private dump). */ export function broadcastVoteUpdate( dumpId: string, voteCount: number, voterId: string, action: "cast" | "remove", onlyUserId?: string, ): void { broadcast( { type: "votes_update", dumpId, voteCount, voterId, action }, onlyUserId ? (client) => client.userId === onlyUserId : undefined, ); } /** * Sends `comment_likes_update` after the route stored a like, to everyone, or * with `onlyUserId` to that user's sockets alone (for a comment on a private * dump). */ export function broadcastCommentLikeUpdate( commentId: string, likeCount: number, likerId: string, action: "cast" | "remove", onlyUserId?: string, ): void { broadcast( { type: "comment_likes_update", commentId, likeCount, likerId, action }, onlyUserId ? (client) => client.userId === onlyUserId : undefined, ); } /** Sends `data` to everyone for a public playlist, else to its owner alone. */ function sendToPlaylistAudience( playlist: Pick, data: ServerToClientMessage, ): void { broadcast( data, (client) => playlist.isPublic || client.userId === playlist.userId, ); } /** * Sends `playlist_created` to the new playlist's audience. The playlist service * calls it. */ export function broadcastPlaylistCreated(playlist: Playlist): void { sendToPlaylistAudience(playlist, { type: "playlist_created", playlist }); } /** * Sends `playlist_updated` to the playlist's audience. `wasPublic` is its * visibility before the change: a playlist that just went private still goes * to everyone, so that the others can drop it. */ export function broadcastPlaylistUpdated( playlist: Playlist, wasPublic: boolean, ): void { sendToPlaylistAudience( { isPublic: playlist.isPublic || wasPublic, userId: playlist.userId }, { type: "playlist_updated", playlist }, ); } /** Sends `playlist_deleted` to the audience the playlist had. */ export function broadcastPlaylistDeleted( playlistId: string, userId: string, isPublic: boolean, ): void { sendToPlaylistAudience({ isPublic, userId }, { type: "playlist_deleted", playlistId, userId, }); } /** * Sends `playlist_dumps_updated` to the playlist's audience, each socket with * only the dumps its user may see, as `GET` would list them. The private dumps * in a public playlist stay hidden from everyone but their posters. */ export function broadcastPlaylistDumpsUpdated( playlist: Playlist, members: PlaylistMember[], ): void { for (const client of clients) { if (!playlist.isPublic && client.userId !== playlist.userId) continue; send(client.socket, { type: "playlist_dumps_updated", playlistId: playlist.id, dumpIds: members .filter((m) => isMemberVisibleTo(m, playlist, client.userId)) .map((m) => m.id), dumpCount: members.length, }); } } /** * Sends everyone `user_updated`, after a profile edit, a role change or an * avatar upload. */ export function broadcastUserUpdated(user: PublicUser): void { broadcast({ type: "user_updated", user }); } /** * Sends everyone `comment_created`. The comments route calls it only for a * comment on a public dump, as it does the two below. */ export function broadcastCommentCreated(comment: Comment): void { broadcast({ type: "comment_created", comment }); } /** Sends everyone `comment_deleted`. */ export function broadcastCommentDeleted( commentId: string, dumpId: string, ): void { broadcast({ type: "comment_deleted", commentId, dumpId }); } /** Sends everyone `comment_updated`. */ export function broadcastCommentUpdated(comment: Comment): void { broadcast({ type: "comment_updated", comment }); } /** * Sends `data` to signed-in sockets only. The chat's history needs a session, * so its live messages do too. */ function sendToMembers(data: ServerToClientMessage): void { broadcast(data, (client) => !!client.userId); } /** * Sends `chat_message` to every signed-in socket, the sender's included. * `nonce` comes from the sender's `chat_send`, and lets it swap its pending * copy for the message. */ export function broadcastChatMessage( message: ChatMessage, nonce?: string, ): void { sendToMembers({ type: "chat_message", message, nonce }); } /** Sends `chat_message_updated` to every signed-in socket. */ export function broadcastChatMessageUpdated(message: ChatMessage): void { sendToMembers({ type: "chat_message_updated", message }); } /** Sends `chat_message_deleted` to every signed-in socket. */ export function broadcastChatMessageDeleted(id: string): void { sendToMembers({ type: "chat_message_deleted", id }); } /** * How long a `chat_typing` true lasts. It must exceed how often the chatbox * resends it (`TYPING_REFRESH`, 3 s), or a steady typist flickers out. */ const TYPING_TTL = 6_000; /** * Returns each user who sent `chat_typing` true within the last * {@link TYPING_TTL}, once, for {@link broadcastTyping}. */ export function getTypingUsers(): OnlineUser[] { const now = Date.now(); const seen = new Map(); for (const client of clients) { if ( client.userId && client.typingAt && now - client.typingAt < TYPING_TTL && !seen.has(client.userId) ) { seen.set(client.userId, { userId: client.userId, username: client.username!, hasAvatar: !!client.avatarMime, avatarVersion: client.avatarVersion, }); } } return Array.from(seen.values()); } /** * Sends `chat_typing_update` to every signed-in socket. The route also calls * it when a typing user's socket closes. */ export function broadcastTyping(): void { sendToMembers({ type: "chat_typing_update", users: getTypingUsers() }); } /** * Records a client's `chat_typing`, and broadcasts the typing list when the * client starts or stops. The route also clears it when the client sends a * message. A client that stops without saying so is expired after * {@link TYPING_TTL}. */ export function setClientTyping(client: WsClient, typing: boolean): void { const wasTyping = !!client.typingAt; client.typingAt = typing ? Date.now() : undefined; // The chatbox resends true every 3 s while typing, which changes nothing. if (typing !== wasTyping) broadcastTyping(); } // Expires the typing users who went quiet without a false, such as by closing // the tab. setInterval(() => { const now = Date.now(); let changed = false; for (const client of clients) { if (client.typingAt && now - client.typingAt >= TYPING_TTL) { client.typingAt = undefined; changed = true; } } if (changed) broadcastTyping(); }, 2_000); /** * Checks whether `userId` is reading the chat on any of their sockets. The * route passes it to `notifyMentions`, which skips those users: they saw the * mention live, and their client chimes for it. */ export function isUserChatOpen(userId: string): boolean { for (const client of clients) { if (client.userId === userId && client.chatOpen) return true; } return false; } export { type RateLimit, takeToken } from "../lib/rate-limit.ts"; /** Records a client's `pong`, which keeps its socket open. */ export function handleClientPong(client: WsClient): void { client.pongReceived = true; } /** * How often every client is pinged. One that hasn't answered the previous ping * is closed. */ const PING_INTERVAL = 30_000; const PING = JSON.stringify({ type: "ping" } satisfies ServerToClientMessage); setInterval(() => { for (const client of clients) { if (client.socket.readyState !== WebSocket.OPEN) { clients.delete(client); continue; } // Undefined, not false, before the first ping. if (client.pongReceived === false) { client.socket.close(1001, "Ping timeout"); clients.delete(client); continue; } client.pongReceived = false; sendRaw(client.socket, PING); } }, PING_INTERVAL);