gerbeur/api/services/ws-service.ts

/**
 * 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<WsClient>();

/** 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<string, OnlineUser>();
  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<Playlist, "isPublic" | "userId">,
  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<string, OnlineUser>();
  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);