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);
}

/**
 * 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.
 */
function broadcastChange<T extends { userId: string }>(
  before: T | null,
  after: T | null,
  isPrivate: (item: T) => boolean,
  messages: {
    created: (item: T) => ServerToClientMessage;
    updated: (item: T) => ServerToClientMessage;
    deleted: (item: T) => ServerToClientMessage;
  },
): void {
  const item = after ?? before;
  if (!item) return;
  const toOwner = (client: WsClient) => client.userId === item.userId;
  const toOthers = (client: WsClient) => client.userId !== item.userId;

  if (!after) {
    broadcast(messages.deleted(item), isPrivate(item) ? toOwner : undefined);
  } else if (!before) {
    broadcast(messages.created(after), isPrivate(after) ? toOwner : undefined);
  } else if (isPrivate(after)) {
    broadcast(messages.updated(after), toOwner);
    if (!isPrivate(before)) broadcast(messages.deleted(after), toOthers);
  } else if (isPrivate(before)) {
    broadcast(messages.updated(after), toOwner);
    broadcast(messages.created(after), toOthers);
  } else {
    broadcast(messages.updated(after));
  }
}

/**
 * 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 `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.
 */
export function broadcastPlaylistChange(
  before: Playlist | null,
  after: Playlist | null,
): void {
  broadcastChange(before, after, (playlist) => !playlist.isPublic, {
    created: (playlist) => ({ type: "playlist_created", playlist }),
    updated: (playlist) => ({ type: "playlist_updated", playlist }),
    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.
 */
export function broadcastPlaylistDumpsUpdated(
  playlist: Pick<Playlist, "id" | "isPublic" | "userId">,
  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 });
}

/**
 * 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<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);
    }
  },
};