gerbeur/api/services/ws-service.ts

/**
 * 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 {
  isMemberVisibleTo,
  type PlaylistMember,
} 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<WsClient>();

/**
 * 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<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 `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<Playlist, "isPublic" | "userId">,
  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<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 `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);
    }
  },
};