gerbeur/api/routes/ws.ts

import { Router } from "@oak/oak";
import { ALLOWED_ORIGINS } from "../config.ts";
import { verifyJWT } from "../lib/jwt.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 { notifyMentions } from "../services/notification-service.ts";
import { hasPermission } from "../lib/permissions.ts";
import {
  castVote,
  getUserVotes,
  removeVote,
} from "../services/vote-service.ts";
import {
  castCommentLike,
  getUserCommentLikes,
  removeCommentLike,
} from "../services/comment-like-service.ts";
import { getUnreadCount } from "../services/notification-service.ts";
import { getUserById } from "../services/user-service.ts";
import {
  APIException,
  type ClientToServerMessage,
} from "../model/interfaces.ts";

const router = new Router();

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

  const token = ctx.request.url.searchParams.get("token");
  const authPayload = token ? await verifyJWT(token) : null;

  const socket = ctx.upgrade();

  // Read the live user so the role reflects the DB, not the (possibly stale)
  // role embedded in the 7-day JWT.
  const user = authPayload ? getUserById(authPayload.userId) : undefined;

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

  // Use addEventListener — more reliable than onopen= with Deno.serve
  socket.addEventListener("open", () => {
    register(client);
    broadcastPresence();

    try {
      const myVotes = authPayload ? getUserVotes(authPayload.userId) : [];
      const myCommentLikes = authPayload
        ? getUserCommentLikes(authPayload.userId)
        : [];
      const unreadNotificationCount = authPayload
        ? getUnreadCount(authPayload.userId)
        : 0;
      const chatRead = authPayload
        ? getChatReadState(authPayload.userId)
        : 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) => {
    let msg: ClientToServerMessage;
    try {
      msg = JSON.parse(event.data as string);
    } catch {
      return;
    }

    switch (msg.type) {
      case "ping":
        socket.send(JSON.stringify({ 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 just dropped.
        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;
    }
  });

  socket.addEventListener("close", () => {
    unregister(client);
    broadcastPresence();
    // If they were mid-compose, drop them from the typing set right away.
    if (client.typingAt) broadcastTyping();
  });
});

function handleVote(
  client: WsClient,
  dumpId: string,
  action: "cast" | "remove",
): void {
  const { socket } = client;

  if (!client.userId) {
    socket.send(
      JSON.stringify({ type: "error", message: "Authentication required" }),
    );
    return;
  }

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

    socket.send(JSON.stringify({
      type: "vote_ack",
      dumpId,
      action,
      success: true,
      voteCount: newCount,
    }));

    broadcastVoteUpdate(dumpId, newCount, client.userId, action);
  } catch (err) {
    const message = err instanceof APIException ? err.message : "Vote failed";
    socket.send(JSON.stringify({ type: "error", message, dumpId, action }));
  }
}

function handleCommentLike(
  client: WsClient,
  commentId: string,
  action: "cast" | "remove",
): void {
  const { socket } = client;

  if (!client.userId) {
    socket.send(
      JSON.stringify({ type: "error", message: "Authentication required" }),
    );
    return;
  }

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

    socket.send(JSON.stringify({
      type: "comment_like_ack",
      commentId,
      action,
      likeCount: newCount,
    }));

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

// Sending, editing and deleting share one bucket: a burst of 8, then one every
// 1.5s — room for a quick back-and-forth, not for a flood.
const CHAT_WRITE_LIMIT: RateLimit = { capacity: 8, refill: 1 / 1.5 };
const TYPING_LIMIT: RateLimit = { capacity: 10, refill: 1 };
// Read markers are cheap for the client to send often; past this they're
// dropped, and the next one catches up.
const CHAT_READ_LIMIT: RateLimit = { capacity: 10, refill: 1 };

function sendChatError(
  client: WsClient,
  message: string,
  nonce?: string,
): void {
  if (client.socket.readyState !== WebSocket.OPEN) return;
  client.socket.send(
    JSON.stringify({ type: "error", scope: "chat", message, nonce }),
  );
}

// Validates the common preconditions of a chat write: signed in and under the
// rate limit. Returns the user id, or undefined after reporting why not.
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;
}

function optionalString(value: unknown, maxLength: number): string | undefined {
  return typeof value === "string" && value.length <= maxLength
    ? value
    : undefined;
}

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),
    );
    // Sending a message ends composition — clear the typing flag before the
    // message broadcast so the indicator doesn't linger next to it.
    setClientTyping(client, false);
    // Broadcast to everyone including the sender, whose pending copy the nonce
    // retires.
    broadcastChatMessage(message, nonce);
    // Skip mention notifications for users who already have the chat open —
    // they've seen the message live.
    notifyMentions(
      userId,
      message.body,
      "chat",
      message.id,
      "",
      undefined,
      isUserChatOpen,
    );
  } catch (err) {
    sendChatError(
      client,
      err instanceof APIException ? err.message : "Failed to send message",
      nonce,
    );
  }
}

// Reads the role from the DB on every moderation check rather than trusting the
// one captured at connect: a demoted moderator loses the power right away, not
// whenever their socket next reconnects.
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");
}

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);
    // A newly added @mention in an edit should still notify (deduped per
    // message+user), skipping anyone already reading the chat.
    notifyMentions(
      userId,
      message.body,
      "chat",
      message.id,
      "",
      undefined,
      isUserChatOpen,
    );
  } catch (err) {
    sendChatError(
      client,
      err instanceof APIException ? err.message : "Failed to edit message",
    );
  }
}

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

// Moves the user's read marker and tells all their sockets, so every tab and
// device drops 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; there's nothing to report.
  }
}

export default router;