/** * 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. Its jobs ping every socket, * closing the ones that stop answering, and expire the typing indicators. * * @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 { type DatedPlaylistMember, dumpFiguresFor, isMemberVisibleTo, } from "../lib/playlist-members.ts"; import type { Job } from "../lib/scheduler.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(); } /** * Records `userId`'s new name on their clients, so presence and the typing * list show it, and rebroadcasts presence. The routes that rename a user call * it, along with `broadcastUserUpdated`. */ export function updateClientUsername(userId: string, username: string): void { for (const client of clients) { if (client.userId === userId) client.username = username; } 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); } /** * Sends `data` to one socket unless it is closing. The WebSocket route answers * a client's own message with it: a pong, a vote's ack, an error. */ export 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` and `notifications_read` 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() }); } /** * Sends the messages that tell each client how an item it may see changed, * for {@link broadcastDumpChange} and {@link broadcastPlaylistChange}. * `before` is null for a new item, `after` for a deleted one. A private item's * changes reach its owner's sockets alone. When it goes private, everyone else * is told it was deleted, and gets nothing of the change; when it goes public, * they are told it was created, since to them it is new. The owner's sockets * and the others' each get a message of their own, built with `toOwner` set * for the former, for what the two don't see alike. */ function broadcastChange( before: T | null, after: T | null, isPrivate: (item: T) => boolean, messages: { created: (item: T, toOwner: boolean) => ServerToClientMessage; updated: (item: T, toOwner: boolean) => ServerToClientMessage; deleted: (item: T) => ServerToClientMessage; }, ): void { const item = after ?? before; if (!item) return; const wasPublic = !!before && !isPrivate(before); const isPublic = !!after && !isPrivate(after); broadcast( !after ? messages.deleted(item) : before ? messages.updated(after, true) : messages.created(after, true), (client) => client.userId === item.userId, ); const toOthers = (client: WsClient) => client.userId !== item.userId; if (after && isPublic) { broadcast( wasPublic ? messages.updated(after, false) : messages.created(after, false), toOthers, ); } else if (wasPublic) { broadcast(messages.deleted(item), toOthers); } } /** * Tells the clients that may see a dump how it changed, as * {@link broadcastChange} says, for the dump service after every write. */ export function broadcastDumpChange( before: Dump | null, after: Dump | null, ): void { broadcastChange(before, after, (dump) => dump.isPrivate, { created: (dump) => ({ type: "dump_created", dump }), updated: (dump) => ({ type: "dump_updated", dump }), deleted: (dump) => ({ type: "dump_deleted", dumpId: dump.id }), }); } /** * Sends `dump_related_changed` after the links between `dump` and each of * `others` changed, for the backlink service. A link shows only to those who * may see both its ends, so a socket is told only when its user may see * `dump` and one of `others`, and of those alone. */ export function broadcastDumpRelatedChanged( dump: Pick, others: Pick[], ): void { for (const client of clients) { const sees = (d: Pick) => !d.isPrivate || d.userId === client.userId; if (!sees(dump)) continue; const seen = others.filter(sees); if (seen.length === 0) continue; send(client.socket, { type: "dump_related_changed", dumpIds: [dump.id, ...seen.map((d) => d.id)], }); } } /** * 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, ); } /** * Tells the clients that may see a playlist how it changed, as * {@link broadcastChange} says, for the playlist service after every write to * its details. A playlist turned private thus leaves the others' views without * their learning what it became. `members` are the dumps of `after`: each * socket gets the playlist with the count of those its user may see and the * date of the latest, the owner's private dumps being theirs alone to count. */ export function broadcastPlaylistChange( before: Playlist | null, after: Playlist | null, members: DatedPlaylistMember[], ): void { const counted = (playlist: Playlist, toOwner: boolean): Playlist => ({ ...playlist, ...dumpFiguresFor(members, playlist, toOwner ? playlist.userId : null), }); broadcastChange(before, after, (playlist) => !playlist.isPublic, { created: (playlist, toOwner) => ({ type: "playlist_created", playlist: counted(playlist, toOwner), }), updated: (playlist, toOwner) => ({ type: "playlist_updated", playlist: counted(playlist, toOwner), }), deleted: (p) => ({ type: "playlist_deleted", playlistId: p.id, userId: p.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, and so * does their number: a client counts the dumps it is sent. */ export function broadcastPlaylistDumpsUpdated( playlist: Pick, members: DatedPlaylistMember[], ): 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), lastDumpAt: dumpFiguresFor(members, playlist, client.userId).lastDumpAt, }); } } /** * 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 }); } /** * Returns the filter of a broadcast about a comment: everyone while its dump * is public, the dump's owner alone while it is private, since no one else may * see it. */ function audienceOf( item: { isPrivate: boolean; userId: string }, ): ((client: WsClient) => boolean) | undefined { return item.isPrivate ? (client) => client.userId === item.userId : undefined; } /** * Sends `comment_created` to those who may see the comment's dump: everyone, * or its owner alone when it is private. The comments route calls it, as it * does the two below. */ export function broadcastCommentCreated( comment: Comment, dump: { isPrivate: boolean; userId: string }, ): void { broadcast({ type: "comment_created", comment }, audienceOf(dump)); } /** Sends `comment_deleted`, as {@link broadcastCommentCreated} does. */ export function broadcastCommentDeleted( commentId: string, dump: { id: string; isPrivate: boolean; userId: string }, ): void { broadcast( { type: "comment_deleted", commentId, dumpId: dump.id }, audienceOf(dump), ); } /** Sends `comment_updated`, as {@link broadcastCommentCreated} does. */ export function broadcastCommentUpdated( comment: Comment, dump: { isPrivate: boolean; userId: string }, ): void { broadcast({ type: "comment_updated", comment }, audienceOf(dump)); } /** * Sends everyone `categories_changed`, after the category manager created, * edited or deleted one, so that every client fetches them again. */ export function broadcastCategoriesChanged(): void { broadcast({ type: "categories_changed" }); } /** * 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, every 2 seconds. `main.ts` starts it. */ export const typingExpiryJob: Job = { name: "typing", everyMs: 2_000, run() { 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(); }, }; /** * 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); /** * Pings every client each {@link PING_INTERVAL}, closing the ones that didn't * answer the previous ping. `main.ts` starts it. */ export const pingJob: Job = { name: "ping", everyMs: PING_INTERVAL, run() { 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); } }, };