gerbeur/api/routes/ws.ts

/**
 * Opens the WebSocket that `WSProvider` keeps for each tab, and handles what
 * the client sends over it. Voting, comment likes and every chat write go
 * through here, not through HTTP. The messages are typed in `shared/ws.ts`,
 * and `services/ws-service.ts` sends the broadcasts.
 *
 * - `GET /ws?token=`, public. A valid session `token` signs the socket in, and
 *   an invalid one leaves it anonymous. Answers 403 to an origin not in
 *   {@link ALLOWED_ORIGINS}, and 426 to a request that isn't an upgrade.
 *
 * @module
 */

import { Router } from "@oak/oak";
import { ALLOWED_ORIGINS } from "../config.ts";
import { authenticate } from "../lib/auth.ts";
import {
  broadcastChatMessage,
  broadcastChatMessageDeleted,
  broadcastChatMessageUpdated,
  broadcastCommentLikeUpdate,
  broadcastPresence,
  broadcastTyping,
  broadcastVoteUpdate,
  getOnlineUsers,
  handleClientPong,
  isUserChatOpen,
  type RateLimit,
  register,
  sendToUser,
  setClientTyping,
  takeToken,
  unregister,
  type WsClient,
} from "../services/ws-service.ts";
import {
  createChatMessage,
  deleteChatMessage,
  getChatReadState,
  markChatRead,
  updateChatMessage,
} from "../services/chat-service.ts";
import {
  getUnreadCount,
  notifyMentions,
} from "../services/notification-service.ts";
import { hasPermission } from "../../shared/permissions.ts";
import {
  castVote,
  getUserVotes,
  removeVote,
} from "../services/vote-service.ts";
import {
  castCommentLike,
  getUserCommentLikes,
  removeCommentLike,
} from "../services/comment-like-service.ts";
import { getUserById } from "../services/user-service.ts";
import { APIException } from "../lib/api-exception.ts";
import type {
  ClientToServerMessage,
  ServerToClientMessage,
} from "../../shared/ws.ts";
import type { User } from "../model/user.ts";

/**
 * Mounted by `api/main.ts` behind the error, Open Graph and CORS middleware.
 * An `APIException` thrown by a handler becomes the JSON error response there.
 */
const router: Router = new Router();

/**
 * Checks whether a socket may be opened from `origin`. Browsers don't apply
 * CORS to WebSockets, so this is what stops another site from opening one with
 * its visitors' tokens.
 */
function isAllowedOrigin(origin: string): boolean {
  return ALLOWED_ORIGINS.includes(origin);
}

router.get("/ws", async (ctx) => {
  const origin = ctx.request.headers.get("origin") ?? "";
  if (!isAllowedOrigin(origin)) {
    ctx.response.status = 403;
    return;
  }

  if (!ctx.isUpgradable) {
    ctx.response.status = 426;
    return;
  }

  // A revoked token (deleted user, changed password) opens an anonymous
  // socket. The client signs out on its next API call, which gets a 401.
  const token = ctx.request.url.searchParams.get("token");
  const authPayload = token ? await authenticate(token) : null;
  let user: User | undefined;
  try {
    user = authPayload ? getUserById(authPayload.userId) : undefined;
  } catch {
    user = undefined;
  }

  const socket = ctx.upgrade();

  const client: WsClient = {
    socket,
    userId: user?.id,
    username: user?.username,
    role: user?.role,
    avatarMime: user?.avatarMime,
  };

  // addEventListener, since `onopen =` proved unreliable under Deno.serve.
  socket.addEventListener("open", () => {
    register(client);
    // The new socket is registered, so it gets this presence_update too,
    // before its welcome.
    broadcastPresence();

    try {
      const myVotes = user ? getUserVotes(user.id) : [];
      const myCommentLikes = user ? getUserCommentLikes(user.id) : [];
      const unreadNotificationCount = user ? getUnreadCount(user.id) : 0;
      const chatRead = user ? getChatReadState(user.id) : undefined;
      socket.send(JSON.stringify({
        type: "welcome",
        users: getOnlineUsers(),
        myVotes,
        myCommentLikes,
        unreadNotificationCount,
        chatReadAt: chatRead?.readAt,
        unreadChatCount: chatRead?.unread ?? 0,
      }));
    } catch (err) {
      console.error("[ws] welcome send failed:", err);
    }
  });

  socket.addEventListener("message", (event) => {
    // An exception escaping a WebSocket listener is uncaught in Deno and ends
    // the process, so nothing a client sends may throw out of here.
    try {
      handleMessage(client, event.data);
    } catch (err) {
      console.error("[ws] message handler failed:", err);
    }
  });

  socket.addEventListener("close", () => {
    unregister(client);
    broadcastPresence();
    // The sweep only sees registered clients, so it would never clear this one.
    if (client.typingAt) broadcastTyping();
  });
});

/**
 * Parses one message from a client and dispatches it by `type`. A message that
 * isn't JSON, or whose `type` is unknown, is ignored. Its fields are still
 * untrusted, so each handler checks their types.
 */
function handleMessage(client: WsClient, data: unknown): void {
  if (typeof data !== "string") return;
  let parsed: unknown;
  try {
    parsed = JSON.parse(data);
  } catch {
    return;
  }
  if (
    typeof parsed !== "object" || parsed === null ||
    typeof (parsed as { type?: unknown }).type !== "string"
  ) return;
  const msg = parsed as ClientToServerMessage;
  const { socket } = client;

  switch (msg.type) {
    case "ping":
      send(socket, { type: "pong" });
      break;
    case "pong":
      handleClientPong(client);
      break;
    case "vote_cast":
      handleVote(client, msg.dumpId, "cast");
      break;
    case "vote_remove":
      handleVote(client, msg.dumpId, "remove");
      break;
    case "comment_like_cast":
      handleCommentLike(client, msg.commentId, "cast");
      break;
    case "comment_like_remove":
      handleCommentLike(client, msg.commentId, "remove");
      break;
    case "chat_send":
      handleChatSend(client, msg.body, msg.replyToId, msg.nonce);
      break;
    case "chat_typing":
      // Typing is cosmetic: past the limit it's dropped, with no error.
      if (
        client.userId &&
        takeToken(`typing:${client.userId}`, TYPING_LIMIT)
      ) {
        setClientTyping(client, msg.typing === true);
      }
      break;
    case "chat_focus":
      client.chatOpen = msg.open === true;
      break;
    case "chat_edit":
      handleChatEdit(client, msg.id, msg.body);
      break;
    case "chat_delete":
      handleChatDelete(client, msg.id);
      break;
    case "chat_read":
      handleChatRead(client, msg.at);
      break;
  }
}

/** Sends `data` unless the socket is closing, where `send` would throw. */
function send(socket: WebSocket, data: ServerToClientMessage): void {
  if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify(data));
}

/**
 * Handles `vote_cast` and `vote_remove`. It acks the voter with `vote_ack` and
 * broadcasts `votes_update`. A failure goes back as an `error` naming the dump
 * and the action, so the client can undo its optimistic vote at once.
 */
function handleVote(
  client: WsClient,
  dumpId: unknown,
  action: "cast" | "remove",
): void {
  const { socket } = client;

  if (!client.userId) {
    send(socket, { type: "error", message: "Authentication required" });
    return;
  }
  if (typeof dumpId !== "string") return;

  try {
    const { voteCount, isPrivate } = action === "cast"
      ? castVote(dumpId, client.userId)
      : removeVote(dumpId, client.userId);

    send(socket, {
      type: "vote_ack",
      dumpId,
      action,
      voteCount,
    });

    // Only a private dump's owner can see it, and so vote on it.
    broadcastVoteUpdate(
      dumpId,
      voteCount,
      client.userId,
      action,
      isPrivate ? client.userId : undefined,
    );
  } catch (err) {
    const message = err instanceof APIException ? err.message : "Vote failed";
    send(socket, { type: "error", message, dumpId, action });
  }
}

/**
 * Handles `comment_like_cast` and `comment_like_remove`, as
 * {@link handleVote} does votes.
 */
function handleCommentLike(
  client: WsClient,
  commentId: unknown,
  action: "cast" | "remove",
): void {
  const { socket } = client;

  if (!client.userId) {
    send(socket, { type: "error", message: "Authentication required" });
    return;
  }
  if (typeof commentId !== "string") return;

  try {
    const { likeCount, isPrivate } = action === "cast"
      ? castCommentLike(commentId, client.userId)
      : removeCommentLike(commentId, client.userId);

    send(socket, {
      type: "comment_like_ack",
      commentId,
      action,
      likeCount,
    });

    broadcastCommentLikeUpdate(
      commentId,
      likeCount,
      client.userId,
      action,
      isPrivate ? client.userId : undefined,
    );
  } catch (err) {
    const message = err instanceof APIException ? err.message : "Like failed";
    send(socket, { type: "error", message, commentId, action });
  }
}

/**
 * Limits each user's chat writes, sends, edits and deletes together, to a burst
 * of 8, then one every 1.5 s. The buckets are per user, so more tabs don't buy
 * more.
 */
const CHAT_WRITE_LIMIT: RateLimit = { capacity: 8, refill: 1 / 1.5 };
const TYPING_LIMIT: RateLimit = { capacity: 10, refill: 1 };
/**
 * Limits each user's `chat_read`. One over the limit is dropped, and the next
 * one catches up, since a marker only moves forward.
 */
const CHAT_READ_LIMIT: RateLimit = { capacity: 10, refill: 1 };

/**
 * Sends a chat-scoped `error`. With the `nonce` of a `chat_send`, the chatbox
 * marks that pending message failed. Without one, it shows `message` as a
 * notice.
 */
function sendChatError(
  client: WsClient,
  message: string,
  nonce?: string,
): void {
  send(client.socket, { type: "error", scope: "chat", message, nonce });
}

/**
 * Returns the id of the user behind a chat write, or undefined after sending
 * the error when they are signed out or over {@link CHAT_WRITE_LIMIT}.
 */
function chatWriter(client: WsClient, nonce?: string): string | undefined {
  if (!client.userId) {
    sendChatError(client, "Authentication required", nonce);
    return undefined;
  }
  if (!takeToken(`chat:${client.userId}`, CHAT_WRITE_LIMIT)) {
    sendChatError(
      client,
      "You're sending messages too fast — wait a moment",
      nonce,
    );
    return undefined;
  }
  return client.userId;
}

/**
 * Returns `value` when it is a string of at most `maxLength`, else undefined.
 */
function optionalString(value: unknown, maxLength: number): string | undefined {
  return typeof value === "string" && value.length <= maxLength
    ? value
    : undefined;
}

/**
 * Handles `chat_send`. It stores the message, broadcasts it, and notifies the
 * @mentioned users who aren't reading the chat.
 */
function handleChatSend(
  client: WsClient,
  body: unknown,
  replyToId: unknown,
  rawNonce: unknown,
): void {
  const nonce = optionalString(rawNonce, 64);
  const userId = chatWriter(client, nonce);
  if (!userId) return;
  if (typeof body !== "string") {
    sendChatError(client, "Message cannot be empty", nonce);
    return;
  }

  try {
    const message = createChatMessage(
      userId,
      body,
      optionalString(replyToId, 64),
    );
    // Before the broadcast, so the typing indicator doesn't linger beside the
    // message.
    setClientTyping(client, false);
    broadcastChatMessage(message, nonce);
    notifyMentions(
      userId,
      message.body,
      "chat",
      message.id,
      "",
      undefined,
      isUserChatOpen,
    );
  } catch (err) {
    sendChatError(
      client,
      err instanceof APIException ? err.message : "Failed to send message",
      nonce,
    );
  }
}

/**
 * Checks whether the client's user may edit and delete others' chat messages.
 * The role is read from the database, not from the socket's connect, so that a
 * demoted moderator loses the right at once.
 */
function canModerateChat(client: WsClient): boolean {
  if (!client.userId) return false;
  try {
    client.role = getUserById(client.userId).role;
  } catch {
    client.role = undefined;
  }
  return !!client.role && hasPermission(client.role, "chat:moderate");
}

/**
 * Handles `chat_edit`. It stores the new body, broadcasts it, and notifies, in
 * the author's name, the users the edit newly mentions.
 */
function handleChatEdit(client: WsClient, id: unknown, body: unknown): void {
  const userId = chatWriter(client);
  if (!userId) return;
  if (typeof id !== "string" || typeof body !== "string") {
    sendChatError(client, "Invalid message");
    return;
  }

  try {
    const message = updateChatMessage(
      id,
      body,
      userId,
      canModerateChat(client),
    );
    broadcastChatMessageUpdated(message);
    // Mentions already notified for this message are deduplicated, so only
    // the new ones notify. A moderator's edit keeps the author's words, so the
    // author mentions.
    notifyMentions(
      message.userId,
      message.body,
      "chat",
      message.id,
      "",
      undefined,
      isUserChatOpen,
    );
  } catch (err) {
    sendChatError(
      client,
      err instanceof APIException ? err.message : "Failed to edit message",
    );
  }
}

/** Handles `chat_delete`. It deletes the message and broadcasts its id. */
function handleChatDelete(client: WsClient, id: unknown): void {
  const userId = chatWriter(client);
  if (!userId) return;
  if (typeof id !== "string") {
    sendChatError(client, "Invalid message");
    return;
  }

  try {
    deleteChatMessage(id, userId, canModerateChat(client));
    broadcastChatMessageDeleted(id);
  } catch (err) {
    sendChatError(
      client,
      err instanceof APIException ? err.message : "Failed to delete message",
    );
  }
}

/**
 * Handles `chat_read`. It moves the user's read marker and sends
 * `chat_read_update` to all their sockets, so every tab and device updates its
 * badge together.
 */
function handleChatRead(client: WsClient, at: unknown): void {
  if (!client.userId || typeof at !== "string") return;
  if (!takeToken(`read:${client.userId}`, CHAT_READ_LIMIT)) return;
  try {
    const state = markChatRead(client.userId, at);
    sendToUser(client.userId, {
      type: "chat_read_update",
      readAt: state.readAt,
      unread: state.unread,
    });
  } catch {
    // A malformed marker changes nothing, and the client doesn't wait for an
    // answer.
  }
}

export default router;