/** * The connected WebSocket clients, and everything pushed to them: presence, * live updates to dumps, comments, playlists and chat, typing indicators, and * the keepalive that drops dead sockets. * * @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"; /** A connected socket, and what the server knows of its user. */ export interface WsClient { /** The socket. */ socket: WebSocket; /** The user's id; undefined for an anonymous socket. */ userId?: string; /** The user's name. */ username?: string; /** The user's role. */ role?: Role; /** The user's avatar type, undefined without one. */ avatarMime?: string; /** When the avatar last changed, to bust the client's cache. */ avatarVersion?: number; /** Whether the client answered the last ping; undefined before the first. */ pongReceived?: boolean; /** Whether this client currently has the chatbox open (is reading chat). */ chatOpen?: boolean; /** * Timestamp (ms) of the client's last "typing" signal; undefined when not * composing. Entries older than {@link TYPING_TTL} are treated as stale. */ typingAt?: number; } /** Every connected client. */ const clients = new Set(); /** Adds a connected client. */ export function register(client: WsClient): void { clients.add(client); } /** Removes a disconnected client. */ export function unregister(client: WsClient): void { clients.delete(client); } /** Records a user's new avatar on their clients and rebroadcasts presence. */ 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 at least one socket, once. */ 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 an already serialized message, if the socket is open. */ function sendRaw(socket: WebSocket, payload: string): void { if (socket.readyState === WebSocket.OPEN) socket.send(payload); } /** Sends one message to one socket. */ function send(socket: WebSocket, data: ServerToClientMessage): void { sendRaw(socket, JSON.stringify(data)); } /** * Sends one message to every client `include` accepts (all by default), * serialized once rather than 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 a message to every socket of `userId`. */ export function sendToUser(userId: string, data: ServerToClientMessage): void { broadcast(data, (client) => client.userId === userId); } /** Logs a deleted user out of every socket and closes them. */ export function disconnectUser(userId: string): void { for (const client of clients) { if (client.userId === userId) { send(client.socket, { type: "force_logout" }); client.socket.close(1000, "Account deleted"); } } } /** * Drops the user's sockets without logging anyone out: each reconnects with * whatever token it holds, and /ws turns a revoked one into a force_logout. * Used on a password change, where the tab that made the change has already * swapped in a fresh token and must stay signed in. */ export function closeUserSockets(userId: string, reason: string): void { for (const client of clients) { if (client.userId === userId) client.socket.close(1000, reason); } } /** Sends everyone the list of online users. */ export function broadcastPresence(): void { broadcast({ type: "presence_update", users: getOnlineUsers() }); } /** Tells everyone a dump was posted. */ export function broadcastNewDump(dump: Dump): void { broadcast({ type: "dump_created", dump }); } /** Tells everyone a dump changed. */ export function broadcastDumpUpdated(dump: Dump): void { broadcast({ type: "dump_updated", dump }); } /** Tells everyone a dump was deleted. */ export function broadcastDumpDeleted(dumpId: string): void { broadcast({ type: "dump_deleted", dumpId }); } /** * Tells everyone a dump's vote count changed. With `onlyUserId`, a private * dump's count goes to that user's sockets alone. */ 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, ); } /** * Tells everyone a comment's like count changed. With `onlyUserId`, a like on * a private dump's comment goes to that user alone. */ 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 a message to everyone for a public playlist, to its owner otherwise. */ function sendToPlaylistAudience( playlist: Pick, data: ServerToClientMessage, ): void { broadcast( data, (client) => playlist.isPublic || client.userId === playlist.userId, ); } /** Tells a new playlist's audience it was created. */ export function broadcastPlaylistCreated(playlist: Playlist): void { sendToPlaylistAudience(playlist, { type: "playlist_created", playlist }); } /** * Tells a playlist's audience it changed. `wasPublic` is the visibility before * this change: a playlist that just went private still goes to everyone, so * non-owners can drop it from their views; one that was and stays private goes * to its owner alone. */ export function broadcastPlaylistUpdated( playlist: Playlist, wasPublic: boolean, ): void { sendToPlaylistAudience( { isPublic: playlist.isPublic || wasPublic, userId: playlist.userId }, { type: "playlist_updated", playlist }, ); } /** Tells a playlist's audience it was deleted. */ export function broadcastPlaylistDeleted( playlistId: string, userId: string, isPublic: boolean, ): void { sendToPlaylistAudience({ isPublic, userId }, { type: "playlist_deleted", playlistId, userId, }); } /** * Tells a playlist's audience its dumps changed. Each viewer gets the members * they can see, as GET would list them: the ids of private dumps don't go out * to the whole audience of a public playlist. */ 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, }); } } /** Tells everyone a user's profile changed. */ export function broadcastUserUpdated(user: PublicUser): void { broadcast({ type: "user_updated", user }); } /** Tells everyone a comment was posted. */ export function broadcastCommentCreated(comment: Comment): void { broadcast({ type: "comment_created", comment }); } /** Tells everyone a comment was deleted. */ export function broadcastCommentDeleted( commentId: string, dumpId: string, ): void { broadcast({ type: "comment_deleted", commentId, dumpId }); } /** Tells everyone a comment was edited. */ export function broadcastCommentUpdated(comment: Comment): void { broadcast({ type: "comment_updated", comment }); } /** * Sends a chat message to signed-in sockets only: history is behind auth, so * the live stream is too — an anonymous socket gets nothing chat-related. */ function sendToMembers(data: ServerToClientMessage): void { broadcast(data, (client) => !!client.userId); } /** * Sends a new chat message to the members. `nonce` is the sender's id for its * pending copy; harmless to everyone else. */ export function broadcastChatMessage( message: ChatMessage, nonce?: string, ): void { sendToMembers({ type: "chat_message", message, nonce }); } /** Tells the members a chat message was edited. */ export function broadcastChatMessageUpdated(message: ChatMessage): void { sendToMembers({ type: "chat_message_updated", message }); } /** Tells the members a chat message was deleted. */ export function broadcastChatMessageDeleted(id: string): void { sendToMembers({ type: "chat_message_deleted", id }); } /** * How long a client's "typing" signal is considered live after its last * refresh. Must exceed the client's typing-ping cadence so a steady typist * never flickers out between pings. */ const TYPING_TTL = 6_000; /** * Returns each user currently composing a chat message (typing signal within * {@link TYPING_TTL}) once, shaped like the online users so the client renders * the same avatars. */ 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 the members the list of users typing. */ export function broadcastTyping(): void { sendToMembers({ type: "chat_typing_update", users: getTypingUsers() }); } /** * Records (or clears) a client's typing state and tells everyone when it * changed. The client also resends "true" periodically while composing; a * sweep expires it if they go quiet without a clean "false" (e.g. they closed * the tab). */ export function setClientTyping(client: WsClient, typing: boolean): void { const wasTyping = !!client.typingAt; client.typingAt = typing ? Date.now() : undefined; // Only rebroadcast on an actual edge to avoid a flood of identical updates // from the periodic typing pings. if (typing !== wasTyping) broadcastTyping(); } // Sweep stale typing entries (clients that stopped pinging without a clean // stop) and rebroadcast when the visible set shrinks. 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 the user has the chatbox open on any of their connected * clients — i.e. they're already reading chat, so a mention notification would * be redundant. */ 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 that a client answered the keepalive ping. */ export function handleClientPong(client: WsClient): void { client.pongReceived = true; } /** * How often every client is pinged. One that hasn't answered the previous ping * is disconnected. */ const PING_INTERVAL = 30_000; /** The serialized ping. */ const PING = JSON.stringify({ type: "ping" } satisfies ServerToClientMessage); setInterval(() => { for (const client of clients) { if (client.socket.readyState !== WebSocket.OPEN) { clients.delete(client); continue; } // Disconnect if no pong since last ping (pongReceived starts undefined, skip first cycle) if (client.pongReceived === false) { client.socket.close(1001, "Ping timeout"); clients.delete(client); continue; } client.pongReceived = false; sendRaw(client.socket, PING); } }, PING_INTERVAL);